"""Agent Soul-backed config assets for skills, files, env, and note text. This service is the single API-side control plane for Agent config assets. It resolves the requested config version (published snapshot, shared draft, or per-user build draft), reads only from ``AgentSoulConfig.config_skills``, ``config_files``, ``env.variables``, and ``config_note``, and mutates only the target Soul JSON. Source file refs are tenant-scoped rather than user-scoped: config writes can be owned by a build-draft Account while the uploaded ``ToolFile`` came from an end-user execution context. It intentionally does not manage object-storage lifecycle: removing or replacing a config asset drops the Soul reference only. """ from __future__ import annotations import io import mimetypes import os.path import urllib.parse import zipfile from collections.abc import Callable from dataclasses import dataclass from enum import StrEnum from operator import itemgetter from typing import Literal from dotenv import dotenv_values from pydantic import BaseModel, ConfigDict, Field from sqlalchemy import select from sqlalchemy.exc import DataError, SQLAlchemyError from sqlalchemy.orm import Session from configs import dify_config from core.app.file_access.controller import DatabaseFileAccessController from core.app.workflow.file_runtime import DifyWorkflowFileRuntime from core.db.session_factory import session_factory as default_session_factory from core.tools.signature import bind_file_uri, sign_tool_file_uri from core.tools.tool_file_manager import ToolFileManager from extensions.ext_storage import storage from factories import file_factory from models.agent import Agent, AgentConfigDraft, AgentConfigDraftType, AgentConfigSnapshot from models.agent_config_entities import ( AgentConfigFileRefConfig, AgentConfigSkillRefConfig, AgentEnvVariableConfig, AgentSoulConfig, validate_config_name, validate_config_skill_name, ) from models.model import UploadFile from models.tools import ToolFile from services.agent.config_skill_normalize_service import ConfigSkillNormalizeService from services.agent.skill_package_service import SkillPackageError from services.skill_management_service import SkillManagementService, SkillManagementServiceError class AgentConfigVersionKind(StrEnum): SNAPSHOT = "snapshot" DRAFT = "draft" BUILD_DRAFT = "build_draft" class AgentConfigMutationSurface(StrEnum): AGENT_STUB = "agent_stub" CONSOLE = "console" class ConfigFileRef(BaseModel): model_config = ConfigDict(extra="forbid") kind: Literal["upload_file", "tool_file"] id: str class AgentConfigServiceError(Exception): """Config operation failure mapped to HTTP status at controller boundaries.""" code: str message: str status_code: int def __init__(self, code: str, message: str, *, status_code: int = 400) -> None: super().__init__(message) self.code = code self.message = message self.status_code = status_code class ConfigPushFileItem(BaseModel): model_config = ConfigDict(extra="forbid") name: str file_ref: ConfigFileRef | None = None class ConfigPushSkillItem(BaseModel): model_config = ConfigDict(extra="forbid") name: str file_ref: ConfigFileRef | None = None class ConfigPushPayload(BaseModel): model_config = ConfigDict(extra="forbid") files: list[ConfigPushFileItem] = Field(default_factory=list) skills: list[ConfigPushSkillItem] = Field(default_factory=list) env_text: str | None = None note: str | None = None @dataclass(slots=True) class AgentConfigTarget: tenant_id: str agent_id: str version_id: str kind: AgentConfigVersionKind writable: bool version: AgentConfigSnapshot | AgentConfigDraft agent_soul: AgentSoulConfig @dataclass(frozen=True, slots=True) class ConfigDownload: filename: str mime_type: str payload: bytes @dataclass(frozen=True, slots=True) class ConfigDownloadRequest: """Short-lived data-plane metadata for one authorized Config asset.""" filename: str mime_type: str size: int download_uri: str class AgentConfigService: """Read and update Agent Soul-backed config assets for one version target.""" PREVIEW_MAX_BYTES = 64 * 1024 def __init__( self, *, tool_file_manager: ToolFileManager | None = None, skill_normalize_service: ConfigSkillNormalizeService | None = None, session_factory: Callable[[], Session] | None = None, ) -> None: self._tool_files = tool_file_manager or ToolFileManager() self._skill_normalizer = skill_normalize_service or ConfigSkillNormalizeService() self._session_factory = session_factory or default_session_factory.create_session def resolve_target( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, user_id: str | None = None, ) -> AgentConfigTarget: with self._session_factory() as session: target = self._resolve_target_in_session( session, tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) return AgentConfigTarget( tenant_id=tenant_id, agent_id=target.agent_id, version_id=target.version_id, kind=target.kind, writable=target.writable, version=target.version, agent_soul=target.agent_soul.model_copy(deep=True), ) def manifest( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, user_id: str | None = None, ) -> dict[str, object]: target = self.resolve_target( tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) return self._manifest_for_target(target) def list_skills( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, user_id: str | None = None, ) -> dict[str, object]: target = self.resolve_target( tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) return { "agent_id": target.agent_id, "config_version": self._config_version_payload(target), "items": self._skill_items_for_target(target, include_runtime_workspace_skills=False), } def list_files( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, user_id: str | None = None, ) -> dict[str, object]: target = self.resolve_target( tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) return { "agent_id": target.agent_id, "config_version": self._config_version_payload(target), "items": [self._serialize_file_item(file_ref) for file_ref in target.agent_soul.config_files], } def request_download( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, kind: Literal["file", "skill"], name: str, user_id: str | None = None, ) -> ConfigDownloadRequest: target = self.resolve_target( tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) if kind == "skill": try: skill = self._require_skill(target.agent_soul, name=name) return self._resolve_download_request( tenant_id=tenant_id, file_kind=skill.file_kind, file_id=self._available_skill_file_id(skill), filename=f"{skill.name}.zip", default_mime_type="application/zip", missing_code="config_skill_not_found", missing_message="config skill payload is missing", ) except AgentConfigServiceError as exc: if exc.code != "config_skill_not_found": raise runtime_skill = next( ( item for item in SkillManagementService().list_runtime_agent_skills( tenant_id=tenant_id, agent_id=agent_id, include_draft=config_version_kind != AgentConfigVersionKind.SNAPSHOT, ) if item["name"] == name ), None, ) if runtime_skill is None: raise AgentConfigServiceError("config_skill_not_found", "config skill not found", status_code=404) return self._resolve_download_request( tenant_id=tenant_id, file_kind="tool_file", file_id=str(runtime_skill["file_id"]), filename=f"{runtime_skill['name']}.zip", default_mime_type="application/zip", missing_code="config_skill_not_found", missing_message="config skill payload is missing", ) file_ref = self._require_file(target.agent_soul, name=name) return self._resolve_download_request( tenant_id=tenant_id, file_kind=file_ref.file_kind, file_id=self._available_file_id(file_ref), filename=file_ref.name, default_mime_type="application/octet-stream", missing_code="config_file_not_found", missing_message="config file payload is missing", ) def pull_skill( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, name: str, user_id: str | None = None, ) -> ConfigDownload: target = self.resolve_target( tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) try: skill = self._require_skill(target.agent_soul, name=name) file_id = self._available_skill_file_id(skill) payload, mime_type = self._load_tool_file_bytes(tenant_id=tenant_id, file_id=file_id) return ConfigDownload( filename=f"{skill.name}.zip", mime_type=mime_type or "application/zip", payload=payload, ) except AgentConfigServiceError as exc: if exc.code != "config_skill_not_found": raise try: result = SkillManagementService().pull_runtime_agent_skill( tenant_id=tenant_id, agent_id=agent_id, name=name, include_draft=config_version_kind != AgentConfigVersionKind.SNAPSHOT, ) return ConfigDownload(filename=result.filename, mime_type=result.mime_type, payload=result.payload) except SkillManagementServiceError as exc: raise AgentConfigServiceError("config_skill_not_found", "config skill not found", status_code=404) from exc def download_skill_url( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, name: str, user_id: str | None = None, ) -> str: result = self.request_download( tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, kind="skill", name=name, user_id=user_id, ) return bind_file_uri(result.download_uri, dify_config.FILES_URL) def inspect_skill( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, name: str, user_id: str | None = None, ) -> dict[str, object]: target = self.resolve_target( tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) try: skill = self._require_skill(target.agent_soul, name=name) file_id = self._available_skill_file_id(skill) archive_bytes, _mime_type = self._load_tool_file_bytes(tenant_id=tenant_id, file_id=file_id) skill_item = self._serialize_skill_item(skill) except AgentConfigServiceError as exc: if exc.code != "config_skill_not_found": raise try: workspace_archive = SkillManagementService().pull_runtime_agent_skill( tenant_id=tenant_id, agent_id=agent_id, name=name, include_draft=config_version_kind != AgentConfigVersionKind.SNAPSHOT, ) except SkillManagementServiceError as skill_exc: raise AgentConfigServiceError( "config_skill_not_found", "config skill not found", status_code=404, ) from skill_exc archive_bytes = workspace_archive.payload skill_item = next( ( item for item in SkillManagementService().list_runtime_agent_skills( tenant_id=tenant_id, agent_id=agent_id, include_draft=config_version_kind != AgentConfigVersionKind.SNAPSHOT, ) if item["name"] == name ), { "id": name, "name": name, "description": "", "size": None, "hash": None, "mime_type": "application/zip", }, ) try: archive_items, skill_md = self._inspect_skill_archive(archive_bytes) except (OSError, ValueError, zipfile.BadZipFile) as exc: raise AgentConfigServiceError( "skill_archive_invalid", "stored config skill archive is invalid", status_code=500, ) from exc return { **skill_item, "source": "config_skill_zip", "files": archive_items, "skill_md": skill_md, "warnings": [], } def preview_skill_file( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, name: str, path: str, user_id: str | None = None, ) -> dict[str, object]: target = self.resolve_target( tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) skill = self._require_skill(target.agent_soul, name=name) file_id = self._available_skill_file_id(skill) member_path = self._normalize_archive_member_path(path) payload = self._load_skill_archive_member( tenant_id=tenant_id, file_id=file_id, path=member_path, ) return self._preview_bytes(path=member_path, size=len(payload), payload=payload) def pull_skill_file( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, name: str, path: str, user_id: str | None = None, ) -> ConfigDownload: target = self.resolve_target( tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) skill = self._require_skill(target.agent_soul, name=name) file_id = self._available_skill_file_id(skill) member_path = self._normalize_archive_member_path(path) payload = self._load_skill_archive_member( tenant_id=tenant_id, file_id=file_id, path=member_path, ) return ConfigDownload( filename=member_path.rsplit("/", 1)[-1], mime_type="application/octet-stream", payload=payload, ) def resolve_skill_file_member_path( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, name: str, path: str, user_id: str | None = None, ) -> str: target = self.resolve_target( tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) skill = self._require_skill(target.agent_soul, name=name) file_id = self._available_skill_file_id(skill) member_path = self._normalize_archive_member_path(path) self._load_skill_archive_member( tenant_id=tenant_id, file_id=file_id, path=member_path, ) return member_path def pull_file( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, name: str, user_id: str | None = None, ) -> ConfigDownload: target = self.resolve_target( tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) file_ref = self._require_file(target.agent_soul, name=name) file_id = self._available_file_id(file_ref) payload, filename, mime_type = self._load_file_ref_bytes( tenant_id=tenant_id, file_kind=file_ref.file_kind, file_id=file_id, ) return ConfigDownload( filename=filename or file_ref.name, mime_type=mime_type or "application/octet-stream", payload=payload ) def download_file_url( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, name: str, user_id: str | None = None, ) -> str: result = self.request_download( tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, kind="file", name=name, user_id=user_id, ) return bind_file_uri(result.download_uri, dify_config.FILES_URL) def upload_skill( self, *, tenant_id: str, agent_id: str, user_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, content: bytes, filename: str, ) -> dict[str, object]: return self._upload_skill( tenant_id=tenant_id, agent_id=agent_id, user_id=user_id, config_version_id=config_version_id, config_version_kind=config_version_kind, content=content, filename=filename, surface=AgentConfigMutationSurface.AGENT_STUB, ) def upload_skill_for_console( self, *, tenant_id: str, agent_id: str, user_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, content: bytes, filename: str, ) -> dict[str, object]: return self._upload_skill( tenant_id=tenant_id, agent_id=agent_id, user_id=user_id, config_version_id=config_version_id, config_version_kind=config_version_kind, content=content, filename=filename, surface=AgentConfigMutationSurface.CONSOLE, ) def _upload_skill( self, *, tenant_id: str, agent_id: str, user_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, content: bytes, filename: str, surface: AgentConfigMutationSurface, ) -> dict[str, object]: with self._session_factory() as session: target = self._resolve_target_in_session( session, tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) self._require_writable(target, surface=surface) try: skill_ref, _package = self._skill_normalizer.normalize( content=content, filename=filename, requested_name=None, tenant_id=tenant_id, user_id=user_id, ) except SkillPackageError as exc: raise AgentConfigServiceError(exc.code, exc.message, status_code=exc.status_code) from exc agent_soul = target.agent_soul.model_copy(deep=True) existing = {item.name: item for item in agent_soul.config_skills} order = [item.name for item in agent_soul.config_skills] if skill_ref.name not in order: order.append(skill_ref.name) existing[skill_ref.name] = skill_ref agent_soul.config_skills = [existing[name] for name in order if name in existing] target.version.config_snapshot = agent_soul session.commit() target.agent_soul = agent_soul return { "skill": self._serialize_skill_item(skill_ref), "config_version": self._config_version_payload(target), } def push( self, *, tenant_id: str, agent_id: str, user_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, payload: ConfigPushPayload, ) -> dict[str, object]: return self._push( tenant_id=tenant_id, agent_id=agent_id, user_id=user_id, config_version_id=config_version_id, config_version_kind=config_version_kind, payload=payload, surface=AgentConfigMutationSurface.AGENT_STUB, ) def push_for_console( self, *, tenant_id: str, agent_id: str, user_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, payload: ConfigPushPayload, ) -> dict[str, object]: return self._push( tenant_id=tenant_id, agent_id=agent_id, user_id=user_id, config_version_id=config_version_id, config_version_kind=config_version_kind, payload=payload, surface=AgentConfigMutationSurface.CONSOLE, ) def push_file_for_console( self, *, tenant_id: str, agent_id: str, user_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, upload_file_id: str, ) -> dict[str, object]: with self._session_factory() as session: target = self._resolve_target_in_session( session, tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) self._require_writable(target, surface=AgentConfigMutationSurface.CONSOLE) upload_file = self._require_console_upload_file_source(session, tenant_id=tenant_id, file_id=upload_file_id) agent_soul = target.agent_soul.model_copy(deep=True) agent_soul.config_files = self._upsert_upload_file( current=agent_soul.config_files, upload_file=upload_file, ) target.version.config_snapshot = agent_soul session.commit() target.agent_soul = agent_soul file_ref = self._require_file(agent_soul, name=upload_file.name) return { "file": self._serialize_file_item(file_ref), "config_version": self._config_version_payload(target), } def _push( self, *, tenant_id: str, agent_id: str, user_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, payload: ConfigPushPayload, surface: AgentConfigMutationSurface, ) -> dict[str, object]: with self._session_factory() as session: target = self._resolve_target_in_session( session, tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) self._require_writable(target, surface=surface) agent_soul = target.agent_soul.model_copy(deep=True) if payload.files: agent_soul.config_files = self._apply_file_updates( session, tenant_id=tenant_id, current=agent_soul.config_files, updates=payload.files, ) if payload.skills: agent_soul.config_skills = self._apply_skill_updates( session, tenant_id=tenant_id, user_id=user_id, current=agent_soul.config_skills, updates=payload.skills, ) if payload.env_text is not None: agent_soul.env.variables = self._apply_env_text(agent_soul.env.variables, payload.env_text) if payload.note is not None: agent_soul.config_note = payload.note target.version.config_snapshot = agent_soul session.commit() target.agent_soul = agent_soul return self._manifest_for_target(target) def preview_file( self, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, name: str, user_id: str | None = None, ) -> dict[str, object]: target = self.resolve_target( tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) file_ref = self._require_file(target.agent_soul, name=name) file_id = self._available_file_id(file_ref) payload, filename, _mime_type = self._load_file_ref_bytes( tenant_id=tenant_id, file_kind=file_ref.file_kind, file_id=file_id, ) return self._preview_bytes( path=filename or file_ref.name, size=file_ref.size, payload=payload, field_name="name" ) def update_env( self, *, tenant_id: str, agent_id: str, user_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, env_text: str, ) -> dict[str, object]: return self._update_env( tenant_id=tenant_id, agent_id=agent_id, user_id=user_id, config_version_id=config_version_id, config_version_kind=config_version_kind, env_text=env_text, surface=AgentConfigMutationSurface.AGENT_STUB, ) def _update_env( self, *, tenant_id: str, agent_id: str, user_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, env_text: str, surface: AgentConfigMutationSurface, ) -> dict[str, object]: with self._session_factory() as session: target = self._resolve_target_in_session( session, tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) self._require_writable(target, surface=surface) agent_soul = target.agent_soul.model_copy(deep=True) agent_soul.env.variables = self._apply_env_text(agent_soul.env.variables, env_text) target.version.config_snapshot = agent_soul session.commit() return {"env_keys": self._env_keys(agent_soul)} def update_note( self, *, tenant_id: str, agent_id: str, user_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, note: str, ) -> dict[str, object]: return self._update_note( tenant_id=tenant_id, agent_id=agent_id, user_id=user_id, config_version_id=config_version_id, config_version_kind=config_version_kind, note=note, surface=AgentConfigMutationSurface.AGENT_STUB, ) def _update_note( self, *, tenant_id: str, agent_id: str, user_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, note: str, surface: AgentConfigMutationSurface, ) -> dict[str, object]: with self._session_factory() as session: target = self._resolve_target_in_session( session, tenant_id=tenant_id, agent_id=agent_id, config_version_id=config_version_id, config_version_kind=config_version_kind, user_id=user_id, ) self._require_writable(target, surface=surface) agent_soul = target.agent_soul.model_copy(deep=True) agent_soul.config_note = note target.version.config_snapshot = agent_soul session.commit() return {"note": agent_soul.config_note} def _resolve_target_in_session( self, session: Session, *, tenant_id: str, agent_id: str, config_version_id: str, config_version_kind: AgentConfigVersionKind, user_id: str | None, ) -> AgentConfigTarget: self._assert_agent_belongs_to_tenant(session, tenant_id=tenant_id, agent_id=agent_id) version: AgentConfigSnapshot | AgentConfigDraft | None try: match config_version_kind: case AgentConfigVersionKind.SNAPSHOT: version = session.scalar( select(AgentConfigSnapshot).where( AgentConfigSnapshot.tenant_id == tenant_id, AgentConfigSnapshot.agent_id == agent_id, AgentConfigSnapshot.id == config_version_id, ) ) writable = False case AgentConfigVersionKind.DRAFT: version = session.scalar( select(AgentConfigDraft).where( AgentConfigDraft.tenant_id == tenant_id, AgentConfigDraft.agent_id == agent_id, AgentConfigDraft.id == config_version_id, AgentConfigDraft.draft_type == AgentConfigDraftType.DRAFT, AgentConfigDraft.account_id.is_(None), ) ) writable = False case AgentConfigVersionKind.BUILD_DRAFT: if not user_id: raise AgentConfigServiceError( "missing_user_id", "user_id is required for build draft config access", status_code=400, ) version = session.scalar( select(AgentConfigDraft).where( AgentConfigDraft.tenant_id == tenant_id, AgentConfigDraft.agent_id == agent_id, AgentConfigDraft.id == config_version_id, AgentConfigDraft.draft_type == AgentConfigDraftType.DEBUG_BUILD, AgentConfigDraft.account_id == user_id, ) ) writable = True case _: raise AgentConfigServiceError("invalid_config_version_kind", "unsupported config version kind") except (DataError, SQLAlchemyError) as exc: session.rollback() raise AgentConfigServiceError( "config_version_not_found", "agent config version was not found", status_code=404, ) from exc if version is None: raise AgentConfigServiceError( "config_version_not_found", "agent config version was not found", status_code=404, ) return AgentConfigTarget( tenant_id=tenant_id, agent_id=agent_id, version_id=version.id, kind=config_version_kind, writable=writable, version=version, agent_soul=AgentSoulConfig.model_validate(version.config_snapshot_dict), ) @staticmethod def _assert_agent_belongs_to_tenant(session: Session, *, tenant_id: str, agent_id: str) -> None: try: found = session.scalar(select(Agent.id).where(Agent.id == agent_id, Agent.tenant_id == tenant_id)) except (DataError, SQLAlchemyError) as exc: session.rollback() raise AgentConfigServiceError( "agent_not_found", "agent does not belong to this tenant", status_code=404 ) from exc if found is None: raise AgentConfigServiceError("agent_not_found", "agent does not belong to this tenant", status_code=404) @staticmethod def _require_writable(target: AgentConfigTarget, *, surface: AgentConfigMutationSurface) -> None: if target.writable: return if surface == AgentConfigMutationSurface.CONSOLE and target.kind == AgentConfigVersionKind.DRAFT: return if surface == AgentConfigMutationSurface.CONSOLE: raise AgentConfigServiceError( "config_not_writable", "config mutations are only allowed for editable drafts", status_code=403, ) raise AgentConfigServiceError( "config_not_writable", "config push is only allowed for build drafts", status_code=403, ) def _apply_file_updates( self, session: Session, *, tenant_id: str, current: list[AgentConfigFileRefConfig], updates: list[ConfigPushFileItem], ) -> list[AgentConfigFileRefConfig]: by_name = {item.name: item for item in current} order = [item.name for item in current] for update in updates: name = validate_config_name(update.name) if update.file_ref is None: by_name.pop(name, None) continue size, hash_value, mime_type = self._validate_source_ref( session, tenant_id=tenant_id, file_ref=update.file_ref, ) if name not in order: order.append(name) by_name[name] = AgentConfigFileRefConfig( name=name, file_kind=update.file_ref.kind, file_id=update.file_ref.id, size=size, hash=hash_value, mime_type=mime_type, ) return [by_name[name] for name in order if name in by_name] @staticmethod def _upsert_upload_file( *, current: list[AgentConfigFileRefConfig], upload_file: UploadFile, ) -> list[AgentConfigFileRefConfig]: name = validate_config_name(upload_file.name) by_name = {item.name: item for item in current} order = [item.name for item in current] if name not in order: order.append(name) by_name[name] = AgentConfigFileRefConfig( name=name, file_kind="upload_file", file_id=upload_file.id, size=upload_file.size, hash=upload_file.hash, mime_type=upload_file.mime_type, ) return [by_name[item_name] for item_name in order if item_name in by_name] def _apply_skill_updates( self, session: Session, *, tenant_id: str, user_id: str, current: list[AgentConfigSkillRefConfig], updates: list[ConfigPushSkillItem], ) -> list[AgentConfigSkillRefConfig]: by_name = {item.name: item for item in current} order = [item.name for item in current] for update in updates: name = validate_config_skill_name(update.name) if update.file_ref is None: by_name.pop(name, None) continue if update.file_ref.kind != "tool_file": raise AgentConfigServiceError( "invalid_skill_file_ref", "config skills must be uploaded as tool files", status_code=400, ) tool_file = self._require_tool_file_source( session, tenant_id=tenant_id, file_id=update.file_ref.id, ) archive_bytes = storage.load_once(tool_file.file_key) try: skill_ref, _package = self._skill_normalizer.normalize( content=archive_bytes, filename=tool_file.name, requested_name=name, tenant_id=tenant_id, user_id=user_id, ) except SkillPackageError as exc: raise AgentConfigServiceError(exc.code, exc.message, status_code=exc.status_code) from exc if name not in order: order.append(name) by_name[name] = skill_ref return [by_name[name] for name in order if name in by_name] def _validate_source_ref( self, session: Session, *, tenant_id: str, file_ref: ConfigFileRef, ) -> tuple[int | None, str | None, str | None]: if file_ref.kind == "tool_file": tool_file = self._require_tool_file_source( session, tenant_id=tenant_id, file_id=file_ref.id, ) return tool_file.size, None, tool_file.mimetype upload_file = self._require_upload_file_source(session, tenant_id=tenant_id, file_id=file_ref.id) return upload_file.size, upload_file.hash, upload_file.mime_type @staticmethod def _require_tool_file_source( session: Session, *, tenant_id: str, file_id: str, ) -> ToolFile: try: tool_file = session.scalar( select(ToolFile).where( ToolFile.id == file_id, ToolFile.tenant_id == tenant_id, ) ) except (DataError, SQLAlchemyError) as exc: session.rollback() raise AgentConfigServiceError( "source_not_found", "source tool file ref is invalid", status_code=404 ) from exc if tool_file is None: raise AgentConfigServiceError( "source_not_found", "source ToolFile not found for this tenant", status_code=404, ) return tool_file @staticmethod def _require_console_upload_file_source(session: Session, *, tenant_id: str, file_id: str) -> UploadFile: try: upload_file = session.scalar( select(UploadFile).where( UploadFile.id == file_id, UploadFile.tenant_id == tenant_id, ) ) except (DataError, SQLAlchemyError) as exc: session.rollback() raise AgentConfigServiceError( "upload_file_not_found", "upload file not found in this workspace", status_code=404, ) from exc if upload_file is None: raise AgentConfigServiceError( "upload_file_not_found", "upload file not found in this workspace", status_code=404, ) return upload_file @staticmethod def _require_upload_file_source(session: Session, *, tenant_id: str, file_id: str) -> UploadFile: try: upload_file = session.scalar( select(UploadFile).where( UploadFile.id == file_id, UploadFile.tenant_id == tenant_id, ) ) except (DataError, SQLAlchemyError) as exc: session.rollback() raise AgentConfigServiceError( "source_not_found", "source upload file ref is invalid", status_code=404 ) from exc if upload_file is None: raise AgentConfigServiceError( "source_not_found", "source UploadFile not found for this tenant", status_code=404, ) return upload_file @staticmethod def _env_keys(agent_soul: AgentSoulConfig) -> list[str]: keys = [item.key or item.name or item.env_name or item.variable for item in agent_soul.env.variables] return [key for key in keys if key] @staticmethod def _apply_env_text(current: list[AgentEnvVariableConfig], env_text: str) -> list[AgentEnvVariableConfig]: updates = AgentConfigService._parse_env_text(env_text) order: list[str] = [] by_key: dict[str, AgentEnvVariableConfig] = {} for item in current: key = AgentConfigService._env_variable_key(item) if key and key not in by_key: by_key[key] = item order.append(key) for key, value in updates: if value is None: by_key.pop(key, None) continue if key not in order: order.append(key) by_key[key] = AgentEnvVariableConfig(key=key, name=key, value=value) return [by_key[key] for key in order if key in by_key] @staticmethod def _env_variable_key(item: AgentEnvVariableConfig) -> str | None: for candidate in (item.key, item.name, item.env_name, item.variable): if isinstance(candidate, str) and candidate.strip(): return candidate.strip() return None @staticmethod def _parse_env_text(env_text: str) -> list[tuple[str, str | None]]: updates: list[tuple[str, str | None]] = [] for raw_line in env_text.splitlines(): line = raw_line.strip() if not line or line.startswith("#"): continue if line.startswith("export "): line = line.removeprefix("export ").strip() if "=" not in line: raise AgentConfigServiceError("invalid_env_file", "env line must contain '='", status_code=400) key, raw_value = line.split("=", 1) key = key.strip() AgentConfigService._validate_env_key(key) if raw_value == "": updates.append((key, None)) continue parsed = dotenv_values(stream=io.StringIO(f"{key}={raw_value}\n")) value = parsed.get(key) if value is None: raise AgentConfigServiceError("invalid_env_file", f"env value for {key} is invalid", status_code=400) updates.append((key, value)) return updates @staticmethod def _validate_env_key(key: str) -> None: normalized = key.strip() if not normalized: raise AgentConfigServiceError("invalid_env_file", "env key must not be blank", status_code=400) if not (normalized[0].isalpha() or normalized[0] == "_"): raise AgentConfigServiceError("invalid_env_file", f"invalid env key {normalized!r}", status_code=400) if any(not (ch.isalnum() or ch == "_") for ch in normalized[1:]): raise AgentConfigServiceError("invalid_env_file", f"invalid env key {normalized!r}", status_code=400) @staticmethod def _manifest_for_target(target: AgentConfigTarget) -> dict[str, object]: return { "agent_id": target.agent_id, "config_version": AgentConfigService._config_version_payload(target), "skills": { "items": AgentConfigService._skill_items_for_target(target, include_runtime_workspace_skills=True) }, "files": { "items": [ AgentConfigService._serialize_file_item(file_ref) for file_ref in target.agent_soul.config_files ] }, "env_keys": AgentConfigService._env_keys(target.agent_soul), "note": target.agent_soul.config_note, } @staticmethod def _skill_items_for_target( target: AgentConfigTarget, *, include_runtime_workspace_skills: bool ) -> list[dict[str, object]]: items = [AgentConfigService._serialize_skill_item(skill) for skill in target.agent_soul.config_skills] if not include_runtime_workspace_skills: return items seen_names = {str(item["name"]) for item in items} for item in SkillManagementService().list_runtime_agent_skills( tenant_id=target.tenant_id, agent_id=target.agent_id, include_draft=target.kind != AgentConfigVersionKind.SNAPSHOT, ): if item["name"] in seen_names: continue seen_names.add(str(item["name"])) items.append(item) return items @staticmethod def _config_version_payload(target: AgentConfigTarget) -> dict[str, object]: return { "id": target.version_id, "kind": target.kind.value, "writable": AgentConfigService._is_console_writable(target), } @staticmethod def _is_console_writable(target: AgentConfigTarget) -> bool: return target.writable or target.kind == AgentConfigVersionKind.DRAFT @staticmethod def _serialize_skill_item(skill: AgentConfigSkillRefConfig) -> dict[str, object]: return { "id": skill.name, "name": skill.name, "file_id": skill.file_id, "is_missing": skill.is_missing, "description": skill.description, "size": skill.size, "hash": skill.hash, "mime_type": skill.mime_type, } @staticmethod def _serialize_file_item(file_ref: AgentConfigFileRefConfig) -> dict[str, object]: return { "id": file_ref.name, "name": file_ref.name, "file_id": file_ref.file_id, "is_missing": file_ref.is_missing, "size": file_ref.size, "hash": file_ref.hash, "mime_type": file_ref.mime_type, } @classmethod def _preview_bytes( cls, *, path: str, size: int | None, payload: bytes, field_name: Literal["name", "path"] = "path", ) -> dict[str, object]: truncated = len(payload) > cls.PREVIEW_MAX_BYTES sample = payload[: cls.PREVIEW_MAX_BYTES] if b"\x00" in sample: return {field_name: path, "size": size, "truncated": truncated, "binary": True, "text": None} try: text = sample.decode("utf-8") except UnicodeDecodeError: if truncated: try: text = sample[:-3].decode("utf-8", errors="strict") except UnicodeDecodeError: return {field_name: path, "size": size, "truncated": truncated, "binary": True, "text": None} else: return {field_name: path, "size": size, "truncated": truncated, "binary": True, "text": None} return {field_name: path, "size": size, "truncated": truncated, "binary": False, "text": text} @classmethod def _normalize_archive_member_path(cls, path: str) -> str: normalized = path.replace("\\", "/").strip("/") if not normalized: raise AgentConfigServiceError( "config_skill_file_invalid", "config skill file path is invalid", status_code=400 ) if "\x00" in normalized or any(ord(ch) < 0x20 for ch in normalized): raise AgentConfigServiceError( "config_skill_file_invalid", "config skill file path is invalid", status_code=400 ) segments = normalized.split("/") if any(not segment or segment in {".", ".."} for segment in segments): raise AgentConfigServiceError( "config_skill_file_invalid", "config skill file path is invalid", status_code=400 ) return "/".join(segments) @classmethod def _inspect_skill_archive(cls, archive_bytes: bytes) -> tuple[list[dict[str, object]], dict[str, object]]: with zipfile.ZipFile(io.BytesIO(archive_bytes)) as archive: members: list[dict[str, object]] = [] directories: set[str] = set() skill_md_preview: dict[str, object] | None = None for info in archive.infolist(): normalized_path = cls._normalize_archive_member_path(info.filename) segments = normalized_path.split("/") directories.update("/".join(segments[:index]) for index in range(1, len(segments))) if info.is_dir(): directories.add(normalized_path) continue members.append( { "path": normalized_path, "name": segments[-1], "type": "file", "previewable": True, "downloadable": True, } ) if normalized_path == "SKILL.md": payload = archive.read(info) skill_md_preview = cls._preview_bytes(path="SKILL.md", size=info.file_size, payload=payload) if skill_md_preview is None or skill_md_preview["binary"]: raise ValueError("skill archive is missing a text SKILL.md") directory_items: list[dict[str, object]] = [ { "path": path, "name": path.rsplit("/", 1)[-1], "type": "directory", "previewable": False, "downloadable": False, } for path in sorted(directories) ] files = sorted([*directory_items, *members], key=itemgetter("path", "type")) return files, skill_md_preview def _load_skill_archive_member(self, *, tenant_id: str, file_id: str, path: str) -> bytes: archive_bytes, _mime_type = self._load_tool_file_bytes(tenant_id=tenant_id, file_id=file_id) normalized_path = self._normalize_archive_member_path(path) try: with zipfile.ZipFile(io.BytesIO(archive_bytes)) as archive: for info in archive.infolist(): candidate_path = self._normalize_archive_member_path(info.filename) if candidate_path != normalized_path: continue if info.is_dir(): raise AgentConfigServiceError( "config_skill_file_not_found", "config skill file not found", status_code=404, ) return archive.read(info) except zipfile.BadZipFile as exc: raise AgentConfigServiceError( "skill_archive_invalid", "stored config skill archive is invalid", status_code=500, ) from exc raise AgentConfigServiceError("config_skill_file_not_found", "config skill file not found", status_code=404) @staticmethod def _require_skill(agent_soul: AgentSoulConfig, *, name: str) -> AgentConfigSkillRefConfig: normalized = validate_config_skill_name(name) for item in agent_soul.config_skills: if item.name == normalized: if item.is_missing or not item.file_id: raise AgentConfigServiceError( "config_skill_missing", "config skill must be uploaded before it can be used", status_code=409, ) return item raise AgentConfigServiceError("config_skill_not_found", "config skill not found", status_code=404) @staticmethod def _require_file(agent_soul: AgentSoulConfig, *, name: str) -> AgentConfigFileRefConfig: normalized = validate_config_name(name) for item in agent_soul.config_files: if item.name == normalized: if item.is_missing or not item.file_id: raise AgentConfigServiceError( "config_file_missing", "config file must be uploaded before it can be used", status_code=409, ) return item raise AgentConfigServiceError("config_file_not_found", "config file not found", status_code=404) @staticmethod def _available_skill_file_id(skill: AgentConfigSkillRefConfig) -> str: if skill.is_missing or not skill.file_id: raise AgentConfigServiceError( "config_skill_missing", "config skill must be uploaded before it can be used", status_code=409, ) return skill.file_id @staticmethod def _available_file_id(file_ref: AgentConfigFileRefConfig) -> str: if file_ref.is_missing or not file_ref.file_id: raise AgentConfigServiceError( "config_file_missing", "config file must be uploaded before it can be used", status_code=409, ) return file_ref.file_id def _load_tool_file_bytes(self, *, tenant_id: str, file_id: str) -> tuple[bytes, str | None]: with self._session_factory() as session: tool_file = session.scalar(select(ToolFile).where(ToolFile.id == file_id, ToolFile.tenant_id == tenant_id)) if tool_file is None: raise AgentConfigServiceError("config_skill_not_found", "config skill payload is missing", status_code=404) return storage.load_once(tool_file.file_key), tool_file.mimetype def _load_file_ref_bytes( self, *, tenant_id: str, file_kind: Literal["upload_file", "tool_file"], file_id: str, ) -> tuple[bytes, str | None, str | None]: with self._session_factory() as session: if file_kind == "tool_file": tool_file = session.scalar( select(ToolFile).where(ToolFile.id == file_id, ToolFile.tenant_id == tenant_id) ) if tool_file is None: raise AgentConfigServiceError( "config_file_not_found", "config file payload is missing", status_code=404 ) return storage.load_once(tool_file.file_key), tool_file.name, tool_file.mimetype upload_file = session.scalar( select(UploadFile).where(UploadFile.id == file_id, UploadFile.tenant_id == tenant_id) ) if upload_file is None: raise AgentConfigServiceError("config_file_not_found", "config file payload is missing", status_code=404) return storage.load_once(upload_file.key), upload_file.name, upload_file.mime_type def _resolve_download_request( self, *, tenant_id: str, file_kind: Literal["upload_file", "tool_file"], file_id: str, filename: str, default_mime_type: str, missing_code: str, missing_message: str, ) -> ConfigDownloadRequest: with self._session_factory() as session: if file_kind == "tool_file": tool_file = session.scalar( select(ToolFile).where(ToolFile.id == file_id, ToolFile.tenant_id == tenant_id) ) if tool_file is None: raise AgentConfigServiceError(missing_code, missing_message, status_code=404) extension = ( os.path.splitext(tool_file.name)[1].lower() or mimetypes.guess_extension(tool_file.mimetype) or os.path.splitext(tool_file.file_key)[1].lower() or ".bin" ) uri = sign_tool_file_uri(tool_file_id=file_id, extension=extension) mime_type = tool_file.mimetype or default_mime_type size = tool_file.size else: upload_file = session.scalar( select(UploadFile).where(UploadFile.id == file_id, UploadFile.tenant_id == tenant_id) ) if upload_file is None: raise AgentConfigServiceError(missing_code, missing_message, status_code=404) runtime = DifyWorkflowFileRuntime(file_access_controller=DatabaseFileAccessController()) uri = runtime.resolve_upload_file_uri(upload_file_id=file_id, as_attachment=True) mime_type = upload_file.mime_type or default_mime_type size = upload_file.size if file_kind == "tool_file": parsed = urllib.parse.urlsplit(uri) query = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True) query.append(("as_attachment", "true")) uri = urllib.parse.urlunsplit(parsed._replace(query=urllib.parse.urlencode(query))) return ConfigDownloadRequest(filename=filename, mime_type=mime_type, size=size, download_uri=uri) @staticmethod def _resolve_download_url( *, tenant_id: str, file_kind: Literal["upload_file", "tool_file"], file_id: str ) -> str | None: controller = DatabaseFileAccessController() from core.app.workflow.file_runtime import DifyWorkflowFileRuntime runtime = DifyWorkflowFileRuntime(file_access_controller=controller) try: if file_kind == "upload_file": return runtime.resolve_upload_file_url( upload_file_id=file_id, for_external=True, as_attachment=True, ) file = file_factory.build_from_mapping( mapping={"transfer_method": "tool_file", "tool_file_id": file_id}, tenant_id=tenant_id, access_controller=controller, ) url = runtime.resolve_file_url(file=file, for_external=True) if not url: return None parsed = urllib.parse.urlsplit(url) query = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True) query.append(("as_attachment", "true")) return urllib.parse.urlunsplit(parsed._replace(query=urllib.parse.urlencode(query))) except ValueError: return None __all__ = [ "AgentConfigMutationSurface", "AgentConfigService", "AgentConfigServiceError", "AgentConfigTarget", "AgentConfigVersionKind", "ConfigDownload", "ConfigDownloadRequest", "ConfigPushFileItem", "ConfigPushPayload", "ConfigPushSkillItem", ]