# retoor 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", "email_account_delete", "email_delete_message", } 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)