# 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_readonly",
"customize_set_css",
"customize_set_js",
"customize_reset",
"project_delete_file",
"delete_project",
"delete_media",
}
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", ""))))
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 == "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 == "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."
)
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,
quota_provider: Any = None,
owner_kind: str = "guest",
owner_id: str = "",
virtual_tools: Any = None,
behavior: 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._fetch = FetchController(settings)
self._docs = DocsController(settings)
self._cost = CostController(quota_provider=quota_provider)
self._chunks = ChunkController(settings)
self._rsearch = RsearchController(settings)
from ..container import ContainerController
self._container = ContainerController(client)
from ..customization import CustomizationController
self._customization = CustomizationController(owner_kind, owner_id)
self._virtual_tools = virtual_tools
self._behavior = behavior
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:
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:
raise AuthRequiredError(
"This information is restricted to administrators.",
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)
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)
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 == "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 == "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:
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)
Reference in New Issue View Git Blame Copy Permalink