dify/api/services/app/console_gateway.py

"""Adapters for app permissions, existing DSL operations and package exporters."""

from __future__ import annotations

import copy
from collections.abc import Callable, Sequence
from contextlib import AbstractContextManager
from dataclasses import dataclass, replace
from typing import Any, BinaryIO, Final, Literal, override
from uuid import UUID, uuid4

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, sessionmaker

from configs import dify_config
from constants.resource_access_token import ResourceAccessTokenResourceType
from core.agent.publish_visibility import agent_has_workflow_callable_active_snapshot
from core.rbac import RBACPermission, RBACResourceScope
from enums import DeploymentEdition
from events.app_event import app_was_updated
from graphon.enums import BuiltinNodeTypes
from graphon.model_runtime.entities.llm_entities import LLMMode
from graphon.model_runtime.entities.message_entities import PromptMessageRole
from libs.datetime_utils import naive_utc_now
from machinery.context import RequestContext
from models.account import Account
from models.agent import Agent, AgentIconType, AgentStatus
from models.model import App, AppMode
from models.tools import WorkflowToolProvider
from repositories.app.agent_app_queries import get_roster_agent_app
from repositories.app.console_repository import ConsoleAppRepository, console_app_actor, require_console_app
from repositories.app.response import app_record
from repositories.resource_access_token_repository import ResourceAccessTokenCleanupRepository
from services.agent.errors import AgentNameConflictError, AgentNotFoundError
from services.agent.home_snapshot_service import AgentHomeSnapshotService
from services.agent.roster_package_entities import RosterAgentPackageExport
from services.agent.roster_package_exporter import RosterAgentPackageExporter
from services.agent.roster_package_importer import RosterAgentPackageImporter
from services.agent.roster_service import AgentRosterService
from services.agent.workspace_service import AgentWorkspaceService
from services.app.access import AppAccessFilter, resolve_app_access_filter
from services.app.agent_app_contracts import AgentAppDetail, AgentAppNameConflictError, AgentAppNotFoundError
from services.app.console_service import (
    AppLifecycle,
    AppPermissions,
    AppTransfers,
    ConsoleAppAccess,
    InvalidAppAccessModesError,
)
from services.app.import_service import AppDefinitionImports
from services.app.response_gateway import AppResponseGateway
from services.app_dsl_service import AppDslService, initialize_imported_app_creator_access
from services.app_import_source import download_app_import_source, try_read_yaml
from services.app_package_service import AppPackageService
from services.app_service import AppService
from services.enterprise import rbac_service
from services.enterprise.enterprise_service import EnterpriseService
from services.entities.app_entities import (
    AppCreationSettings,
    AppDeletion,
    AppEvent,
    AppExportOptions,
    AppListParams,
    AppRecord,
    AppUpdateArguments,
    CreateAppParams,
    ImportedAppPackage,
    UpdateAppParams,
)
from services.entities.dsl_entities import (
    AppImportPackage,
    AppImportParams,
    CheckDependenciesResult,
    DraftGuardedImportParams,
    Import,
    ImportStatus,
    PackagedWorkflowTool,
)
from services.errors.base import NoPermissionError
from services.feature_service import FeatureService
from services.rbac import contracts as rbac_contracts
from services.rbac.members import MemberService
from services.recommended_app_package_service import RecommendedAppPackageService
from services.system_feature_service import SystemFeatureService
from services.tools.workflow_tools_manage_service import WorkflowToolManageService
from services.workflow_service import WorkflowService
from services.workflow_tool_references import WorkflowToolTarget
from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task

# Positions as the console editor lays out its templates (START_INITIAL_POSITION and
# NODE_WIDTH_X_OFFSET in web/app/components/workflow/constants.ts).
_START_POSITION: Final = (80, 282)
_NODE_X_STEP: Final = 300
_DISCARDED_IMPORT_STATUSES: Final = frozenset({ImportStatus.FAILED})
# A copy that waits for version confirmation keeps nothing.
_DISCARDED_COPY_STATUSES: Final = frozenset({ImportStatus.FAILED, ImportStatus.PENDING})


def _editor_node(node_id: str, column: int, data: dict[str, Any]) -> dict[str, Any]:
    """A node as the editor's generateNewNode shapes it."""
    x, y = _START_POSITION
    return {
        "id": node_id,
        "type": "custom",
        "data": data,
        "position": {"x": x + _NODE_X_STEP * column, "y": y},
        "targetPosition": "left",
        "sourcePosition": "right",
    }


def _editor_edge(source: str, target: str) -> dict[str, Any]:
    return {
        "id": f"{source}-{target}",
        "source": source,
        "sourceHandle": "source",
        "target": target,
        "targetHandle": "target",
    }


# The Chatflow template the console editor saves on first open (use-workflow-template.ts).
_CHATFLOW_STARTER_GRAPH: Final[dict[str, Any]] = {
    "nodes": [
        _editor_node("start", 0, {"type": BuiltinNodeTypes.START, "title": "Start", "variables": []}),
        _editor_node(
            "llm",
            1,
            {
                "type": BuiltinNodeTypes.LLM,
                "title": "LLM",
                "model": {"provider": "", "name": "", "mode": LLMMode.CHAT, "completion_params": {"temperature": 0.7}},
                "prompt_template": [{"role": PromptMessageRole.SYSTEM, "text": ""}],
                "context": {"enabled": False, "variable_selector": []},
                "vision": {"enabled": False},
                "memory": {
                    "window": {"enabled": False, "size": 10},
                    "query_prompt_template": "{{#sys.query#}}\n\n{{#sys.files#}}",
                },
            },
        ),
        _editor_node(
            "answer",
            2,
            {"type": BuiltinNodeTypes.ANSWER, "title": "Answer", "variables": [], "answer": "{{#llm.text#}}"},
        ),
    ],
    "edges": [_editor_edge("start", "llm"), _editor_edge("llm", "answer")],
}

# The draft a new app starts with. A workflow starts empty: its entry (Start or a trigger) is chosen while building.
_STARTER_DRAFT_GRAPHS: Final[dict[AppMode, dict[str, Any]]] = {
    AppMode.WORKFLOW: {"nodes": [], "edges": []},
    AppMode.ADVANCED_CHAT: _CHATFLOW_STARTER_GRAPH,
}
_STARTER_DRAFT_FEATURES: Final[dict[str, Any]] = {"retriever_resource": {"enabled": True}}


@dataclass(frozen=True)
class ConsoleAppPermissions(AppPermissions):
    snapshot: rbac_contracts.MyPermissionsResponse
    access_filter: AppAccessFilter

    @override
    def filter_list(self, params: AppListParams) -> AppListParams:
        params = params.model_copy(deep=True)
        self.access_filter.apply_to_params(params)
        return params

    @override
    def keys_for(self, app_ids: list[str]) -> dict[str, list[str]]:
        return self.snapshot.app.permission_keys_by_resource_ids(app_ids)


class EnterpriseConsoleAppAccess(ConsoleAppAccess):
    @override
    def require_import(self, context: RequestContext, kind: Literal["url", "dsl", "agent"]) -> None:
        if not dify_config.RBAC_ENABLED:
            return
        permissions = (
            (RBACPermission.AGENT_CREATE, RBACPermission.AGENT_IMPORT_EXPORT_DSL)
            if kind == "agent"
            else (RBACPermission.APP_IMPORT_EXPORT_DSL, RBACPermission.AGENT_IMPORT_EXPORT_DSL)
            if kind == "url"
            else (RBACPermission.APP_IMPORT_EXPORT_DSL,)
        )
        checks = (
            rbac_service.RBACService.CheckAccess.check(
                context.active_workspace_id, context.account_id, scene=permission, resource_type=None, resource_id=None
            )
            for permission in permissions
        )
        allowed = any(checks) if kind == "url" else all(checks)
        if not allowed:
            raise NoPermissionError("You do not have permission to import this App")

    @override
    def imported_permissions(self, context: RequestContext, app_id: str) -> list[str]:
        return self.created_permissions(context, app_id) if dify_config.RBAC_ENABLED else []

    @override
    def initialize_import_access(self, app_id: str) -> None:
        if SystemFeatureService.is_webapp_auth_enabled():
            EnterpriseService.WebAppAuth.update_app_access_mode(app_id, "private")

    @override
    def inherit_access(self, source_app_id: str, app_id: str) -> None:
        if not SystemFeatureService.is_webapp_auth_enabled():
            return
        EnterpriseService.WebAppAuth.inherit_app_access_mode(source_app_id, app_id)

    def __init__(self, *, members: MemberService) -> None:
        self._members = members

    @override
    def permissions(self, context: RequestContext, *, app_id: str | None = None) -> ConsoleAppPermissions:
        permissions = self._members.permissions(context.active_workspace_id, context.account_id, app_id=app_id)
        access_filter = AppAccessFilter.unrestricted()
        if dify_config.RBAC_ENABLED and app_id is None:
            access_filter = resolve_app_access_filter(
                context.active_workspace_id, context.account_id, permissions=permissions
            )
        return ConsoleAppPermissions(permissions, access_filter)

    @override
    def created_permissions(self, context: RequestContext, app_id: str) -> list[str]:
        keys = self._members.resource_permissions(
            context.active_workspace_id, context.account_id, rbac_contracts.RBACResourceType.APP, [app_id]
        )
        return keys.get(app_id, [])

    @override
    def initialize_created_app(self, context: RequestContext, app_id: str) -> None:
        if dify_config.RBAC_ENABLED:
            rbac_service.RBACService.AppAccess.replace_whitelist(
                context.active_workspace_id,
                context.account_id,
                app_id,
                rbac_contracts.ReplaceMemberBindings(automatic_include_workspace_members=True),
            )
            initialize_created_app_rbac_access_task.delay(
                context.active_workspace_id, context.account_id, app_id=app_id
            )

    @override
    def access_modes(self, app_ids: list[str]) -> dict[str, str]:
        if not SystemFeatureService.is_webapp_auth_enabled():
            return {}
        settings = EnterpriseService.WebAppAuth.batch_get_app_access_mode_by_id(app_ids=app_ids)
        if len(settings) != len(app_ids):
            raise InvalidAppAccessModesError("Invalid app id in webapp auth")
        return {app_id: setting.access_mode for app_id, setting in settings.items()}

    @override
    def access_mode(self, app_id: str) -> str | None:
        if not SystemFeatureService.is_webapp_auth_enabled():
            return None
        return EnterpriseService.WebAppAuth.get_app_access_mode_by_id(app_id=app_id).access_mode

    @override
    def can_export_version(self, workspace_id: str) -> bool:
        return (
            dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD
            or FeatureService.get_workspace_plan(workspace_id).is_paid
        )


class AppTransferGateway(AppTransfers, AppDefinitionImports):
    def __init__(
        self,
        *,
        session_factory: sessionmaker[Session],
        dsl_factory: Callable[[Session], AppDslService],
        packages: AppPackageService,
        agent_packages: RosterAgentPackageExporter,
        agent_importer: RosterAgentPackageImporter,
        recommended_packages: RecommendedAppPackageService,
    ) -> None:
        self._session_factory = session_factory
        self._dsl_factory = dsl_factory
        self._packages = packages
        self._agent_packages = agent_packages
        self._agent_importer = agent_importer
        self._recommended_packages = recommended_packages

    @override
    def download_import(self, url: str) -> AbstractContextManager[BinaryIO]:
        return download_app_import_source(url)

    @override
    def read_import_yaml(self, source: BinaryIO) -> str | None:
        return try_read_yaml(source)

    @override
    def read_app_package(self, source: BinaryIO) -> AppImportPackage | None:
        return self._packages.read_package(source)

    def _import_actor(self, context: RequestContext) -> Account:
        with self._session_factory() as session:
            return console_app_actor(session, context)

    @override
    def import_dsl(
        self,
        context: RequestContext,
        params: AppImportParams,
        *,
        package: AppImportPackage | None = None,
        preserve_agent_bindings: bool = False,
    ) -> Import:
        return self._import(
            context,
            params,
            package=package,
            preserve_agent_bindings=preserve_agent_bindings,
            discarded=_DISCARDED_IMPORT_STATUSES,
        )

    @override
    def import_copy(self, context: RequestContext, params: AppImportParams) -> Import:
        return self._import(
            context, params, package=None, preserve_agent_bindings=True, discarded=_DISCARDED_COPY_STATUSES
        )

    def _import(
        self,
        context: RequestContext,
        params: AppImportParams,
        *,
        package: AppImportPackage | None,
        preserve_agent_bindings: bool,
        discarded: frozenset[ImportStatus],
    ) -> Import:
        account = self._import_actor(context)
        with self._session_factory() as session:
            result = self._dsl_factory(session).import_app(
                account=account,
                import_mode=params.mode,
                yaml_content=params.yaml_content,
                yaml_url=params.yaml_url,
                name=params.name,
                description=params.description,
                icon_type=params.icon_type,
                icon=params.icon,
                icon_background=params.icon_background,
                app_id=params.app_id,
                package=package,
                preserve_agent_bindings=preserve_agent_bindings,
                draft_hash=params.draft_hash if isinstance(params, DraftGuardedImportParams) else None,
            )
            if result.status in discarded:
                session.rollback()
            else:
                session.commit()
        initialize_imported_app_creator_access(
            result, tenant_id=context.active_workspace_id, account_id=context.account_id
        )
        return result

    @override
    def confirm_import(self, context: RequestContext, import_id: str) -> tuple[Import, bool]:
        result = self.confirm_definition(context, import_id)
        return result, result.app_created

    @override
    def confirm_definition(self, context: RequestContext, import_id: str) -> Import:
        account = self._import_actor(context)
        with self._session_factory() as session:
            result = self._dsl_factory(session).confirm_import(import_id=import_id, account=account)
            if result.status == ImportStatus.FAILED:
                session.rollback()
            else:
                session.commit()
        initialize_imported_app_creator_access(
            result, tenant_id=context.active_workspace_id, account_id=context.account_id
        )
        return result

    @override
    def import_agent_package(self, context: RequestContext, source: BinaryIO) -> ImportedAppPackage:
        account = self._import_actor(context)
        result = self._agent_importer.import_package(
            source=source,
            tenant_id=context.active_workspace_id,
            account=account,
        )
        return ImportedAppPackage(result.app_id, result.agent_id, result.warnings)

    @override
    def import_agent_template(self, context: RequestContext, params: AppImportParams) -> ImportedAppPackage:
        assert params.template_id is not None
        assert params.version_id is not None
        source = self._recommended_packages.get_source(app_id=str(params.template_id), version_id=params.version_id)
        result = self._agent_importer.import_template(
            source=source,
            tenant_id=context.active_workspace_id,
            account=self._import_actor(context),
            name=params.name,
            description=params.description,
            icon_type=params.icon_type,
            icon=params.icon,
            icon_background=params.icon_background,
        )
        return ImportedAppPackage(result.app_id, result.agent_id, result.warnings)

    @override
    def import_agent_package_url(self, context: RequestContext, params: AppImportParams) -> ImportedAppPackage:
        assert params.package_url is not None
        result = self._agent_importer.import_package_url(
            url=params.package_url,
            tenant_id=context.active_workspace_id,
            account=self._import_actor(context),
            name=params.name,
            description=params.description,
            icon_type=params.icon_type,
            icon=params.icon,
            icon_background=params.icon_background,
        )
        return ImportedAppPackage(result.app_id, result.agent_id, result.warnings)

    @override
    def check_dependencies(self, context: RequestContext, app_id: str) -> CheckDependenciesResult:
        return AppDslService.check_app_dependencies(tenant_id=context.active_workspace_id, app_id=app_id)

    @override
    def plan_workflow_tools(
        self, context: RequestContext, tools: Sequence[PackagedWorkflowTool], *, creates_root_app: bool
    ) -> dict[str, WorkflowToolTarget]:
        """Reuse tools that already exist here, and admit, name and reserve ids for the ones to create."""
        workspace_id = context.active_workspace_id
        with self._session_factory() as session:
            account = console_app_actor(session, context)
            existing = {
                provider.id: provider.name
                for provider in session.scalars(
                    select(WorkflowToolProvider).where(
                        WorkflowToolProvider.tenant_id == workspace_id,
                        WorkflowToolProvider.id.in_([tool.provider_id for tool in tools]),
                    )
                )
            }
            taken_names = set(
                session.scalars(select(WorkflowToolProvider.name).where(WorkflowToolProvider.tenant_id == workspace_id))
            )
            can_manage_tools = account.is_admin_or_owner
        missing = [tool for tool in tools if tool.provider_id not in existing]
        if missing and dify_config.RBAC_ENABLED:
            can_manage_tools = rbac_service.RBACService.CheckAccess.check(
                workspace_id, context.account_id, scene=RBACPermission.TOOL_MANAGE, resource_type=None, resource_id=None
            )
        if missing and not can_manage_tools:
            raise NoPermissionError("You do not have permission to create the workflow tools this App calls")
        AppDslService.check_app_quota(tenant_id=workspace_id, new_apps=len(missing) + int(creates_root_app))
        targets = {
            provider_id: WorkflowToolTarget(provider_id=provider_id, name=name, create=False)
            for provider_id, name in existing.items()
        }
        for tool in missing:
            name, suffix = tool.name, 2
            while name in taken_names:
                name, suffix = f"{tool.name}_{suffix}", suffix + 1
            taken_names.add(name)
            targets[tool.provider_id] = WorkflowToolTarget(provider_id=str(uuid4()), name=name, create=True)
        return targets

    @override
    def create_workflow_tool(
        self, context: RequestContext, *, app_id: str, tool: PackagedWorkflowTool, target: WorkflowToolTarget
    ) -> None:
        """Publish an imported workflow and expose it as the Workflow Tool its callers already reference."""
        with self._session_factory.begin() as session:
            account = console_app_actor(session, context)
            app = require_console_app(session, context, app_id)
            workflow = WorkflowService().publish_workflow(session=session, app_model=app, account=account)
            app.workflow_id = workflow.id
            app.updated_by = account.id
            app.updated_at = naive_utc_now()
        WorkflowToolManageService.create_workflow_tool(
            user_id=context.account_id,
            tenant_id=context.active_workspace_id,
            workflow_app_id=app_id,
            name=target.name,
            label=tool.label,
            icon=dict(tool.icon),
            description=tool.description,
            parameters=tool.parameters,
            privacy_policy=tool.privacy_policy,
            labels=tool.labels,
            import_id=target.provider_id,
        )

    @override
    def export_dsl(self, context: RequestContext, app_id: str, options: AppExportOptions) -> str:
        with self._session_factory() as session:
            app = require_console_app(session, context, app_id)
            dsl = self._dsl_factory(session)
            prepared = dsl.load_export_data(
                app_model=app,
                session=session,
                include_secret=options.include_secret,
                workflow_id=options.workflow_id,
                version_id=options.version_id,
                preserve_agent_bindings=options.preserve_agent_bindings,
            )
        return dsl.serialize_export_data(prepared)

    @override
    def export_app_package(
        self, context: RequestContext, app_id: str, options: AppExportOptions
    ) -> RosterAgentPackageExport:
        with self._session_factory() as session:
            app = require_console_app(session, context, app_id)
        return self._packages.export_app(
            app_model=app,
            include_secret=options.include_secret,
            workflow_id=options.workflow_id,
            preserve_agent_bindings=options.preserve_agent_bindings,
            may_bundle=lambda called_app: self._may_export_app(context, called_app),
        )

    @staticmethod
    def _may_export_app(context: RequestContext, app: App) -> bool:
        """Apply the export endpoint's App check to a workflow the exported App calls."""
        # Without RBAC, the workspace role the endpoint admitted covers every App in the workspace.
        if not dify_config.RBAC_ENABLED or app.maintainer == context.account_id:
            return True
        return rbac_service.RBACService.CheckAccess.check(
            context.active_workspace_id,
            context.account_id,
            scene=RBACPermission.APP_IMPORT_EXPORT_DSL,
            resource_type=RBACResourceScope.APP,
            resource_id=app.id,
        )

    @override
    def export_agent_package(
        self, *, workspace_id: str, agent_id: str, version_id: UUID | None
    ) -> RosterAgentPackageExport:
        return self._agent_packages.export(tenant_id=workspace_id, agent_id=agent_id, version_id=version_id)


def notify_agent_app_updated(tenant_id: str, app_id: str) -> None:
    """Notify existing listeners after an Agent App change has committed."""
    app_was_updated.send(AppEvent(app_id, tenant_id, str(AppMode.AGENT)))


class AppLifecycleGateway(AppLifecycle):
    """Integrate App persistence with existing Agent transaction participants.

    Agent services retain their current API. Their database-only operations join
    the App transaction; notification and resource cleanup run after it closes.
    """

    def __init__(self, *, session_factory: sessionmaker[Session]) -> None:
        self._session_factory = session_factory

    @override
    def create(self, context: RequestContext, params: CreateAppParams, settings: AppCreationSettings) -> AppRecord:
        with self._session_factory.begin() as session:
            account = console_app_actor(session, context)
            app = ConsoleAppRepository.insert_app_record(
                context.active_workspace_id, params, account, settings, session=session
            )
            if app.mode == AppMode.AGENT:
                initial_soul = AppService.prepare_agent_soul(context.active_workspace_id, session=session)
                try:
                    AgentRosterService(session).create_backing_agent_for_app(
                        tenant_id=context.active_workspace_id,
                        account_id=context.account_id,
                        app_id=app.id,
                        name=params.name,
                        description=params.description or "",
                        role=params.agent_role,
                        icon_type=AgentIconType(params.icon_type) if params.icon_type else None,
                        icon=params.icon,
                        icon_background=params.icon_background,
                        initial_soul=initial_soul,
                    )
                except IntegrityError as exc:
                    raise AgentNameConflictError() from exc
            return app_record(app, session=session, projection="detail-with-site")

    @override
    def create_draft(self, context: RequestContext, app_id: str) -> None:
        with self._session_factory() as session:
            app = require_console_app(session, context, app_id)
            WorkflowService(self._session_factory).sync_draft_workflow(
                app_model=app,
                graph=copy.deepcopy(_STARTER_DRAFT_GRAPHS[AppMode(app.mode)]),
                features=copy.deepcopy(_STARTER_DRAFT_FEATURES),
                unique_hash=None,
                account=console_app_actor(session, context),
                environment_variables=[],
                conversation_variables=[],
                session=session,
            )

    @override
    def delete(self, context: RequestContext, app_id: str) -> AppDeletion:
        with self._session_factory.begin() as session:
            app = require_console_app(session, context, app_id)
            return self._delete_in_session(context, app, session=session)

    def delete_agent(self, context: RequestContext, agent_id: str) -> AppDeletion:
        """Resolve and delete an active roster Agent App in one transaction."""
        with self._session_factory.begin() as session:
            app = get_roster_agent_app(session, tenant_id=context.active_workspace_id, agent_id=agent_id)
            if app is None:
                raise AgentAppNotFoundError
            ResourceAccessTokenCleanupRepository(session=session).delete_resource_relations(
                tenant_id=context.active_workspace_id,
                resource_type=ResourceAccessTokenResourceType.APP,
                resource_id=app.id,
            )
            return self._delete_in_session(context, app, session=session)

    def get_agent_detail(self, context: RequestContext, agent_id: str) -> AgentAppDetail:
        """Materialize the runtime App and Agent while the legacy session is open."""
        with self._session_factory() as session:
            roster = AgentRosterService(session)
            try:
                # Workflow-only Agents may still need their legacy hidden runtime App created here.
                app = roster.get_agent_runtime_app_model(tenant_id=context.active_workspace_id, agent_id=agent_id)
            except AgentNotFoundError as exc:
                raise AgentAppNotFoundError from exc
            agent = session.scalar(
                select(Agent).where(
                    Agent.tenant_id == context.active_workspace_id,
                    Agent.id == agent_id,
                    Agent.status == AgentStatus.ACTIVE,
                )
            )
            if agent is None:
                raise AgentAppNotFoundError
            record = app_record(app, session=session, projection="detail-with-site")
            # The legacy response view detects old Agent configurations here,
            # including its existing mode update and commit.
            should_mask_agent_mode = bool(
                record.app_model_config is not None and app.is_agent_with_session(session=session)
            )
            return AgentAppDetail(
                app=record,
                agent_id=agent.id,
                app_id=agent.app_id,
                backing_app_id=roster.runtime_backing_app_id(agent),
                hidden_app_backed=bool(agent.backing_app_id and agent.backing_app_id != agent.app_id),
                role=agent.role or "",
                access_ready=agent_has_workflow_callable_active_snapshot(session=session, agent=agent),
                should_mask_agent_mode=should_mask_agent_mode,
            )

    def get_agent_build_conversation(self, context: RequestContext, agent_id: str, *, app_id: str) -> tuple[str, int]:
        """Commit the caller's Build conversation if the Agent still owns the same runtime App."""
        with self._session_factory.begin() as session:
            roster = AgentRosterService(session)
            agent = session.scalar(
                select(Agent).where(
                    Agent.tenant_id == context.active_workspace_id,
                    Agent.id == agent_id,
                    Agent.status == AgentStatus.ACTIVE,
                )
            )
            if agent is None or roster.runtime_backing_app_id(agent) != app_id:
                raise AgentAppNotFoundError
            try:
                conversation_id = roster.get_or_create_build_conversation(
                    tenant_id=context.active_workspace_id,
                    agent_id=agent_id,
                    account_id=context.account_id,
                    commit=False,
                )
            except AgentNotFoundError as exc:
                raise AgentAppNotFoundError from exc
            count = roster.count_agent_app_debug_conversation_messages(conversation_id=conversation_id)
            return conversation_id, count

    def update_agent(self, context: RequestContext, agent_id: str, params: UpdateAppParams) -> str:
        """Commit updates for the tenant's active roster Agent App and return its id."""
        with self._session_factory() as session:
            app = get_roster_agent_app(session, tenant_id=context.active_workspace_id, agent_id=agent_id)
            if app is None:
                raise AgentAppNotFoundError
            args: AppUpdateArguments = {
                "name": params.name,
                "description": params.description,
                "icon_type": params.icon_type,
                "icon": params.icon,
                "icon_background": params.icon_background,
                "use_icon_as_answer_icon": params.use_icon_as_answer_icon,
                "max_active_requests": params.max_active_requests,
                "role": params.role,
            }
            try:
                return ConsoleAppRepository.update_app(app, args, account_id=context.account_id, session=session).id
            except AgentNameConflictError as exc:
                raise AgentAppNameConflictError from exc

    def agent_updated(self, context: RequestContext, app_id: str) -> None:
        """Notify listeners after the Agent App update has committed."""
        notify_agent_app_updated(context.active_workspace_id, app_id)

    @staticmethod
    def _delete_in_session(context: RequestContext, app: App, *, session: Session) -> AppDeletion:
        """Stage App deletion and resource retirement in the caller's transaction."""
        deleted = ConsoleAppRepository.delete_app_record(app, account_id=context.account_id, session=session)
        for binding_id in deleted.binding_ids:
            AgentWorkspaceService.retire_binding(
                session=session, tenant_id=context.active_workspace_id, binding_id=binding_id
            )
        snapshots = (
            AgentHomeSnapshotService.retire_all_for_agent(
                session=session, tenant_id=context.active_workspace_id, agent_id=deleted.backing_agent_id
            )
            if deleted.backing_agent_id is not None
            else []
        )
        workspaces = AgentWorkspaceService.retire_all_for_app(
            session=session, tenant_id=context.active_workspace_id, app_id=app.id
        )
        return replace(deleted, home_snapshot_ids=snapshots, workspace_ids=workspaces)

    @override
    def prepare_creation(self, context: RequestContext, params: CreateAppParams) -> AppCreationSettings:
        return AppService.prepare_creation(context.active_workspace_id, params)

    @override
    def created(self, context: RequestContext, app: AppRecord) -> None:
        AppService.notify_created_app(
            event=AppEvent(app.id, context.active_workspace_id, app.mode_compatible_with_agent),
            account_id=context.account_id,
            backing_agent_id=app.bound_agent_id,
        )

    @override
    def updated(self, context: RequestContext, app: AppRecord) -> None:
        app_was_updated.send(AppEvent(app.id, context.active_workspace_id, app.mode_compatible_with_agent))

    @override
    def deleted(self, context: RequestContext, deleted: AppDeletion) -> None:
        AppService.notify_deleted_app(deleted, account_id=context.account_id)

    @override
    def present(self, context: RequestContext, app: AppRecord, *, mask_credentials: bool = False) -> AppRecord:
        app = replace(
            app, deleted_tools=AppResponseGateway.find_deleted_tools(context.active_workspace_id, app.tool_references)
        )
        return AppResponseGateway.mask_record(context, app) if mask_credentials else app


class AgentAppDetailPresentationGateway:
    """Resolve Agent detail metadata without holding the App persistence session."""

    def __init__(self, *, members: MemberService) -> None:
        self._members = members

    def present(self, context: RequestContext, detail: AgentAppDetail) -> AgentAppDetail:
        """Enrich a detached App record with access mode, permissions and tool availability."""
        access_mode = (
            EnterpriseService.WebAppAuth.get_app_access_mode_by_id(app_id=detail.app.id).access_mode
            if SystemFeatureService.is_webapp_auth_enabled()
            else None
        )
        app = replace(
            detail.app,
            access_mode=access_mode,
            deleted_tools=AppResponseGateway.find_deleted_tools(
                context.active_workspace_id, detail.app.tool_references
            ),
        )
        app = (
            AppResponseGateway.mask_agent_mode(context, app)
            if detail.should_mask_agent_mode
            else AppResponseGateway.mask_record(context, app)
        )
        permissions = self._members.permissions(
            context.active_workspace_id, context.account_id, agent_id=detail.agent_id
        )
        keys = permissions.agent.permission_keys_by_resource_ids([detail.agent_id]).get(detail.agent_id, [])
        return replace(detail, app=replace(app, permission_keys=keys))