devplacepy/devplacepy/services/devii/actions/dispatcher.py

# retoor <retoor@molodetz.nl>

from __future__ import annotations

import json
import logging
import re
from typing import Any
from urllib.parse import quote, urlencode

from ..config import Settings
from ..errors import (
    AuthRequiredError,
    DeviiError,
    ToolInputError,
    error_result,
    unexpected_result,
)
from ..agentic.controller import AgenticController
from ..agentic.state import record_mutation
from ..avatar import AvatarController
from ..chunks import ChunkController, get_store, serve_resource, wrap_if_large
from ..cost import CostController
from ..docs import DocsController
from ..fetch import FetchController
from ..http_client import PlatformClient
from ..rsearch import RsearchController
from ..tasks.controller import TaskController
from ..text import format_response
from .spec import Action, Catalog

MUTATING_METHODS = ("POST", "DELETE", "PUT", "PATCH")

CONFIRM_REQUIRED = {
    "project_set_private",
    "project_set_readonly",
    "customize_set_css",
    "customize_set_js",
    "customize_reset",
    "project_delete_file",
    "delete_project",
    "delete_media",
    "regenerate_api_key",
    "regenerate_avatar",
    "delete_post",
    "delete_comment",
    "delete_gist",
    "delete_attachment",
    "delete_issue_attachment",
    "delete_comment_attachment",
    "admin_media_purge",
    "admin_delete_news",
    "admin_reset_all_ai_quota",
    "admin_reset_guest_ai_quota",
    "admin_reset_user_ai_quota",
    "backup_delete",
    "backup_schedule_delete",
    "notification_reset",
    "gateway_provider_delete",
    "gateway_model_delete",
    "gateway_quota_rule_delete",
    "email_account_delete",
    "email_delete_message",
    "game_prestige",
}

CONDITIONAL_CONFIRM = {
    "container_instance_action",
    "container_exec",
}

DESTRUCTIVE_COMMAND = re.compile(
    r"(?:^|[\s;&|`(])(?:sudo\s+)?(rm|rmdir|unlink|shred|truncate|wipefs|mkfs\S*|dd)(?:\s|$)"
    r"|>\s*/"
    r"|\bdrop\s+(?:table|database)\b"
    r"|\bdelete\s+from\b"
    r"|\bgit\s+clean\b"
    r"|\s-delete\b",
    re.IGNORECASE,
)

logger = logging.getLogger("devii.dispatch")


def _is_confirmed(arguments: dict[str, Any]) -> bool:
    return str(arguments.get("confirm", "")).strip().lower() in (
        "true",
        "1",
        "yes",
        "on",
    )


def _is_destructive_command(arguments: dict[str, Any]) -> bool:
    return bool(DESTRUCTIVE_COMMAND.search(str(arguments.get("command", ""))))


_DEVII_MECHANIC_EVENTS = {
    "update_behavior": "devii.behavior.update",
    "tool_create": "devii.tool.create",
    "tool_update": "devii.tool.update",
    "tool_delete": "devii.tool.delete",
    "create_task": "devii.task.create",
    "update_task": "devii.task.update",
    "delete_task": "devii.task.delete",
    "reflect": "devii.lesson.reflect",
    "forget_lessons": "devii.lesson.forget",
    "customize_set_css": "devii.customization.css.set",
    "customize_set_js": "devii.customization.js.set",
    "customize_reset": "devii.customization.reset",
    "notification_set": "devii.notification.set",
    "notification_reset": "devii.notification.reset",
    "email_account_set": "email.account.set",
    "email_account_delete": "email.account.delete",
    "email_send": "email.send",
    "email_delete_message": "email.message.delete",
    "email_move_message": "email.message.move",
    "email_mark": "email.message.flag",
    "email_set_flags": "email.message.flag",
    "telegram_send": "telegram.send",
    "interactions_set": "profile.interactions",
}

_DEVII_CONTAINER_EVENTS = {
    "container_create_instance": "container.instance.create",
    "container_configure_instance": "container.instance.configure",
    "container_exec": "container.instance.exec",
    "container_schedule": "container.schedule.create",
}


def _safe_args(arguments: dict[str, Any]) -> dict[str, Any]:
    out: dict[str, Any] = {}
    for key, value in arguments.items():
        text = str(value)
        out[key] = text if len(text) <= 200 else text[:200] + "..."
    return out


def confirmation_error(name: str, arguments: dict[str, Any]) -> ToolInputError | None:
    if _is_confirmed(arguments):
        return None
    if name in ("customize_set_css", "customize_set_js"):
        scope = str(arguments.get("scope", "")).strip() or "(unspecified)"
        return ToolInputError(
            "Before saving, confirm the scope with the user: should this apply to THIS page type "
            f"('{scope}') or to the whole site ('global')? Ask them, then call again with the chosen "
            "scope and confirm=true."
        )
    if name == "customize_reset":
        return ToolInputError(
            "Deleting customizations cannot be undone. Ask the user to confirm, then call again with "
            "confirm=true."
        )
    if name == "notification_reset":
        return ToolInputError(
            "Resetting clears every notification preference and restores the platform defaults; it "
            "cannot be undone. Ask the user to confirm, then call again with confirm=true."
        )
    if name == "project_set_private":
        value = str(arguments.get("value", "")).strip()
        return ToolInputError(
            f"Setting a project {'private' if value == 'true' else 'public'} "
            "hides it from (or exposes it to) other members. Ask the user to confirm this "
            "change explicitly, then call again with confirm=true."
        )
    if name == "project_set_readonly":
        return ToolInputError(
            "Setting a project read-only makes every file immutable and blocks all further "
            "edits. Ask the user to confirm this explicitly first, then call again with "
            "confirm=true."
        )
    if name == "project_delete_file":
        path = str(arguments.get("path", "")).strip() or "(unspecified)"
        return ToolInputError(
            f"Deleting '{path}' is permanent and cannot be undone. Show the user the exact path, "
            "get explicit confirmation, then call again with confirm=true."
        )
    if name == "delete_project":
        return ToolInputError(
            "Deleting a project permanently removes the project and ALL of its files. Ask the user "
            "to confirm this explicitly first, then call again with confirm=true."
        )
    if name == "delete_media":
        return ToolInputError(
            "Deleting media removes it from every display surface, including its parent post or "
            "project. Show the user the exact media, get explicit confirmation, then call again "
            "with confirm=true."
        )
    if name == "regenerate_api_key":
        return ToolInputError(
            "Regenerating your API key immediately invalidates the current key, which may break "
            "scripts or services that use it. Ask the user to confirm explicitly, then call again "
            "with confirm=true."
        )
    if name == "delete_post":
        slug = str(arguments.get("slug", arguments.get("post_slug", ""))).strip()
        return ToolInputError(
            f"Deleting the post '{slug}' is permanent and removes it from the site. "
            "Ask the user to confirm explicitly, then call again with confirm=true."
        )
    if name == "delete_comment":
        uid = str(arguments.get("uid", "")).strip() or "(unspecified)"
        return ToolInputError(
            f"Deleting comment '{uid}' is permanent and cannot be undone. "
            "Ask the user to confirm explicitly, then call again with confirm=true."
        )
    if name == "delete_gist":
        uid = str(arguments.get("uid", "")).strip() or "(unspecified)"
        return ToolInputError(
            f"Deleting gist '{uid}' is permanent. Show the user the exact gist, "
            "get explicit confirmation, then call again with confirm=true."
        )
    if name == "delete_attachment":
        uid = str(arguments.get("uid", "")).strip() or "(unspecified)"
        return ToolInputError(
            f"Deleting attachment '{uid}' is permanent and removes it from its parent. "
            "Ask the user to confirm explicitly, then call again with confirm=true."
        )
    if (
        name == "container_instance_action"
        and str(arguments.get("action", "")).strip().lower() == "delete"
    ):
        return ToolInputError(
            "Deleting a container instance is irreversible and destroys its state. Ask the user to "
            "confirm explicitly, then call again with confirm=true."
        )
    if name == "container_exec" and _is_destructive_command(arguments):
        command = str(arguments.get("command", "")).strip()
        return ToolInputError(
            "This command can permanently delete files or data (it contains a destructive operation "
            f"such as rm, dd, truncate, or drop): {command!r}. Show the user the exact command, get "
            "explicit confirmation, then call again with confirm=true."
        )
    if name == "email_account_delete":
        label = str(arguments.get("account", "")).strip() or "(unspecified)"
        return ToolInputError(
            f"Deleting the saved email account '{label}' removes its stored connection settings. "
            "Ask the user to confirm explicitly, then call again with confirm=true."
        )
    if name == "email_delete_message":
        uid = str(arguments.get("uid", "")).strip() or "(unspecified)"
        return ToolInputError(
            f"Deleting message {uid} on the remote mailbox is permanent unless a Trash folder is "
            "given. Show the user the exact message, get explicit confirmation, then call again "
            "with confirm=true."
        )
    if name in CONFIRM_REQUIRED:
        return ToolInputError(
            "This removes the item as a soft delete: it disappears from every surface and is only "
            "restorable from the admin trash. Show the user exactly what will be deleted, get "
            "explicit confirmation, then call again with confirm=true."
        )
    return None


class Dispatcher:
    def __init__(
        self,
        catalog: Catalog,
        client: PlatformClient,
        settings: Settings,
        tasks: TaskController,
        agentic: AgenticController,
        avatar: AvatarController | None = None,
        browser: Any = None,
        is_admin: bool = False,
        is_primary_admin: bool = False,
        quota_provider: Any = None,
        owner_kind: str = "guest",
        owner_id: str = "",
        virtual_tools: Any = None,
        behavior: Any = None,
        interaction: Any = None,
    ) -> None:
        self._actions = catalog.by_name()
        self._client = client
        self._settings = settings
        self._tasks = tasks
        self._agentic = agentic
        self._avatar = avatar
        self._browser = browser
        self._is_admin = is_admin
        self._is_primary_admin = is_primary_admin
        self._owner_kind = owner_kind
        self._owner_id = owner_id
        self._fetch = FetchController(settings)
        self._docs = DocsController(settings, is_admin=is_admin)
        self._cost = CostController(quota_provider=quota_provider)
        self._chunks = ChunkController(settings)
        self._rsearch = RsearchController(settings, owner_kind, owner_id)
        from ..container import ContainerController

        self._container = ContainerController(client, owner_id=owner_id)
        from ..customization import CustomizationController

        self._customization = CustomizationController(owner_kind, owner_id)
        from ..notification import NotificationController

        self._notification = NotificationController(owner_kind, owner_id)
        from ..ai_correction import AiCorrectionController

        self._ai_correction = AiCorrectionController(owner_kind, owner_id)
        from ..ai_modifier import AiModifierController

        self._ai_modifier = AiModifierController(owner_kind, owner_id)
        from ..email import EmailController

        self._email = EmailController(settings, owner_kind, owner_id)
        from ..telegram import TelegramSendController

        self._telegram = TelegramSendController(owner_kind, owner_id)
        self._virtual_tools = virtual_tools
        self._behavior = behavior
        self._interaction = interaction
        self._read_files: set[tuple[str, str]] = set()

    @staticmethod
    def _file_key(arguments: dict[str, Any]) -> tuple[str, str] | None:
        from devplacepy.project_files import (
            normalize_path,
            ProjectFileError as _PFError,
        )

        raw = arguments.get("path")
        if not raw:
            return None
        try:
            path = normalize_path(raw)
        except _PFError:
            return None
        return str(arguments.get("project_slug", "")), path

    def is_read_only(self, name: str) -> bool:
        action = self._actions.get(name)
        return bool(action and action.is_read_only)

    async def dispatch(self, name: str, arguments: dict[str, Any]) -> str:
        action = self._actions.get(name)
        if action is None:
            if self._virtual_tools is not None and self._virtual_tools.has(name):
                logger.info("Dispatch virtual tool %s args=%s", name, list(arguments))
                try:
                    return await self._virtual_tools.run(name, arguments)
                except DeviiError as exc:
                    return error_result(exc)
                except Exception as exc:  # noqa: BLE001 - surfaced to the model as data
                    logger.exception("Virtual tool %s crashed", name)
                    return unexpected_result(exc)
            return error_result(ToolInputError(f"Unknown tool: {name}"))

        logger.info("Dispatch %s args=%s", name, list(arguments))
        try:
            if action.requires_auth and not self._client.authenticated:
                self._audit_denied(name, "authentication required", arguments)
                raise AuthRequiredError(
                    "Not authenticated. Ask the user for credentials and call the login tool first.",
                    tool=name,
                )
            if action.requires_admin and not self._is_admin:
                self._audit_denied(name, "administrator access required", arguments)
                raise AuthRequiredError(
                    "This information is restricted to administrators.",
                    tool=name,
                )
            if action.requires_primary_admin and not self._is_primary_admin:
                self._audit_denied(name, "primary administrator access required", arguments)
                raise AuthRequiredError(
                    "This tool is restricted to the primary administrator.",
                    tool=name,
                )
            guard = confirmation_error(name, arguments)
            if guard is not None:
                raise guard
            resource_key = self._resource_key(action, arguments)
            if resource_key:
                cached = serve_resource(resource_key, self._settings.max_response_chars)
                if cached is not None:
                    logger.info("Resource cache hit for %s (%s)", name, resource_key)
                    return cached
            result = await self._run(action, arguments)
            self._audit_mechanic(action, arguments)
            if action.handler == "chunks":
                return result
            return wrap_if_large(
                result, self._settings.max_response_chars, resource_key
            )
        except DeviiError as exc:
            logger.info("Dispatch %s failed: %s", name, exc.message)
            return error_result(exc)
        except Exception as exc:  # noqa: BLE001 - surfaced to the model as data
            logger.exception("Dispatch %s crashed", name)
            return unexpected_result(exc)

    def _audit_denied(self, name: str, reason: str, arguments: dict[str, Any]) -> None:
        from devplacepy.services.audit import record as audit

        actor_kind = "user" if self._owner_kind == "user" else self._owner_kind
        audit.record_system(
            "security.authz.denied",
            actor_kind=actor_kind,
            actor_uid=self._owner_id if self._owner_kind == "user" else None,
            actor_role="admin" if self._is_admin else (actor_kind if actor_kind != "user" else "member"),
            origin="devii",
            via_agent=1,
            result="denied",
            summary=f"Devii denied {name}: {reason}",
            metadata={"tool": name, "reason": reason, "args": _safe_args(arguments)},
        )

    def _audit_mechanic(self, action: Action, arguments: dict[str, Any]) -> None:
        from devplacepy.services.audit import record as audit

        name = action.name
        event_key = _DEVII_MECHANIC_EVENTS.get(name)
        if event_key is None and action.handler == "container":
            if name == "container_instance_action":
                event_key = f"container.instance.{arguments.get('action', 'action')}"
            else:
                event_key = _DEVII_CONTAINER_EVENTS.get(name)
        if event_key is None:
            return
        actor_kind = "user" if self._owner_kind == "user" else self._owner_kind
        audit.record_system(
            event_key,
            actor_kind=actor_kind,
            actor_uid=self._owner_id if self._owner_kind == "user" else None,
            actor_role="admin" if self._is_admin else (actor_kind if actor_kind != "user" else "member"),
            origin="devii",
            via_agent=1,
            summary=f"Devii {name} for {self._owner_id or self._owner_kind}",
            metadata={"tool": name, "args": _safe_args(arguments)},
        )

    async def _run(self, action: Action, arguments: dict[str, Any]) -> str:
        if action.handler == "status":
            return json.dumps(
                {
                    "authenticated": self._client.authenticated,
                    "user": self._client.username,
                },
                ensure_ascii=False,
            )

        if action.handler == "login":
            result = await self._client.login(
                email=self._require(arguments, "email"),
                password=self._require(arguments, "password"),
                remember_me=str(arguments.get("remember_me", "on")).lower()
                not in ("", "false", "off", "no"),
            )
            return json.dumps(result, ensure_ascii=False)

        if action.handler == "logout":
            return json.dumps(await self._client.logout(), ensure_ascii=False)

        if action.handler == "task":
            return await self._tasks.dispatch(action.name, arguments)

        if action.handler == "agentic":
            return await self._agentic.dispatch(action.name, arguments)

        if action.handler == "fetch":
            return await self._fetch.dispatch(action.name, arguments)

        if action.handler == "docs":
            return await self._docs.dispatch(action.name, arguments)

        if action.handler == "cost":
            return await self._cost.dispatch(action.name, arguments)

        if action.handler == "chunks":
            return await self._chunks.dispatch(action.name, arguments)

        if action.handler == "rsearch":
            return await self._rsearch.dispatch(action.name, arguments)

        if action.handler == "container":
            return await self._container.dispatch(action.name, arguments)

        if action.handler == "customization":
            return await self._customization.dispatch(action.name, arguments)

        if action.handler == "notification":
            return await self._notification.dispatch(action.name, arguments)

        if action.handler == "ai_correction":
            return await self._ai_correction.dispatch(action.name, arguments)

        if action.handler == "ai_modifier":
            return await self._ai_modifier.dispatch(action.name, arguments)

        if action.handler == "email":
            return await self._email.dispatch(action.name, arguments)

        if action.handler == "telegram":
            return await self._telegram.dispatch(action.name, arguments)

        if action.handler == "behavior":
            if self._behavior is None:
                return error_result(
                    ToolInputError(
                        "Behavior configuration is not available in this context."
                    )
                )
            return await self._behavior.dispatch(action.name, arguments)

        if action.handler == "interaction":
            if self._interaction is None:
                return error_result(
                    ToolInputError(
                        "Interactive prompts are not available in this context."
                    )
                )
            return await self._interaction.dispatch(action.name, arguments)

        if action.handler == "virtual_tool":
            if self._virtual_tools is None:
                return error_result(
                    ToolInputError(
                        "User-defined tools are not available in this context."
                    )
                )
            return await self._virtual_tools.dispatch(action.name, arguments)

        if action.handler == "avatar":
            if self._avatar is None:
                return error_result(
                    ToolInputError(
                        "No avatar is attached; devii's on-screen actions need the web interface."
                    )
                )
            return await self._avatar.dispatch(action.name, arguments)

        if action.handler == "client":
            if self._browser is None:
                return error_result(
                    ToolInputError(
                        "No browser is attached; client-side actions need the web interface."
                    )
                )
            return await self._browser.dispatch(action.name, arguments)

        return await self._run_http(action, arguments)

    def _build_request(
        self, action: Action, arguments: dict[str, Any]
    ) -> tuple[str, dict[str, Any], dict[str, Any], tuple[str, str] | None]:
        url_path = action.path
        params: dict[str, Any] = {}
        data: dict[str, Any] = {}
        file_field: tuple[str, str] | None = None

        for param in action.params:
            if param.name not in arguments or arguments[param.name] is None:
                if param.required:
                    raise ToolInputError(
                        f"Missing required parameter '{param.name}' for {action.name}."
                    )
                continue
            value = arguments[param.name]
            if param.location == "path":
                url_path = url_path.replace(
                    "{" + param.name + "}", quote(str(value), safe="")
                )
            elif param.location == "query":
                params[param.name] = value
            elif param.location == "body":
                data[param.name] = value
            elif param.location == "file":
                file_field = (param.name, str(value))

        if action.freeform_body:
            extra = arguments.get("form_fields") or {}
            if isinstance(extra, dict):
                data.update({str(k): v for k, v in extra.items()})

        return url_path, params, data, file_field

    def _resource_key(self, action: Action, arguments: dict[str, Any]) -> str | None:
        try:
            if action.handler == "fetch" and action.name == "fetch_url":
                url = str(arguments.get("url", "")).strip()
                if not url:
                    return None
                if "://" not in url:
                    url = "https://" + url
                return f"fetch:{url}"
            if action.handler == "http" and action.method == "GET":
                url_path, params, _, _ = self._build_request(action, arguments)
                query = urlencode(sorted((str(k), str(v)) for k, v in params.items()))
                return f"http:GET {url_path}?{query}"
        except ToolInputError:
            return None
        return None

    async def _file_exists(self, arguments: dict[str, Any]) -> bool:
        slug = str(arguments.get("project_slug", "")).strip()
        path = str(arguments.get("path", "")).strip()
        if not slug or not path:
            return False
        response = await self._client.call(
            method="GET",
            path=f"/projects/{quote(slug, safe='')}/files/raw",
            params={"path": path},
            headers={"X-Requested-With": "fetch"},
        )
        return response.status_code == 200

    async def _run_http(self, action: Action, arguments: dict[str, Any]) -> str:
        if action.name == "project_write_file":
            key = self._file_key(arguments)
            if (
                key is not None
                and key not in self._read_files
                and await self._file_exists(arguments)
            ):
                raise ToolInputError(
                    f"Read '{key[1]}' before overwriting it. It already exists; call "
                    "project_read_file first. For an existing file prefer the line tools "
                    "(project_replace_lines, project_insert_lines, project_delete_lines, "
                    "project_append_file); project_write_file replaces the entire file."
                )

        url_path, params, data, file_field = self._build_request(action, arguments)

        headers = {"X-Requested-With": "fetch"} if action.ajax else None
        response = await self._client.call(
            method=action.method,
            path=url_path,
            params=params or None,
            data=data or None,
            file_field=file_field,
            headers=headers,
        )
        if action.name in (
            "project_read_file",
            "project_read_lines",
            "project_write_file",
        ):
            key = self._file_key(arguments)
            if key is not None:
                self._read_files.add(key)
        if action.method in MUTATING_METHODS and not action.is_read_only:
            record_mutation(action.name)
            store = get_store()
            if store is not None:
                store.invalidate_resources(prefix="http:")
        return format_response(response)

    @staticmethod
    def _require(arguments: dict[str, Any], key: str) -> str:
        value = arguments.get(key)
        if value is None or str(value).strip() == "":
            raise ToolInputError(f"Missing required field '{key}'.")
        return str(value)