devplacepy/devplacepy/services/devii/interaction/capabilities.py

# retoor <retoor@molodetz.nl>

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

CHANNEL_SITE_CHAT = "site-chat"
CHANNEL_TELEGRAM = "telegram"
CHANNEL_CLI = "cli"
CHANNEL_API = "api"
CHANNEL_UNKNOWN = "unknown"

UI_TOOL_NAMES = frozenset(
    {
        "ui_prompt",
        "ui_cancel",
        "ui_notify",
        "interactions_get",
        "interactions_set",
    }
)

DEFAULT_LIMITS = {
    "max_options": 32,
    "max_fields": 24,
    "max_label_chars": 200,
    "max_help_chars": 500,
    "callback_data_bytes": 0,
}

_SITE_CAPS = {
    "rich_widgets": True,
    "markdown": True,
    "confirm": True,
    "choice_single": True,
    "choice_multi": True,
    "text_input": True,
    "number_input": True,
    "date_input": True,
    "file_input": False,
    "inline_buttons": True,
    "polls": False,
    "web_app": False,
    "streaming_partial": True,
    "ephemeral_status": True,
}

_TELEGRAM_CAPS = {
    "rich_widgets": False,
    "markdown": True,
    "confirm": True,
    "choice_single": True,
    "choice_multi": True,
    "text_input": True,
    "number_input": True,
    "date_input": True,
    "file_input": False,
    "inline_buttons": True,
    "polls": True,
    "web_app": False,
    "streaming_partial": False,
    "ephemeral_status": False,
}

_PLAIN_CAPS = {
    "rich_widgets": False,
    "markdown": True,
    "confirm": True,
    "choice_single": True,
    "choice_multi": True,
    "text_input": True,
    "number_input": True,
    "date_input": True,
    "file_input": False,
    "inline_buttons": False,
    "polls": False,
    "web_app": False,
    "streaming_partial": False,
    "ephemeral_status": False,
}

_EMPTY_CAPS = {key: False for key in _SITE_CAPS}

_MATRICES: dict[str, dict[str, bool]] = {
    CHANNEL_SITE_CHAT: _SITE_CAPS,
    CHANNEL_TELEGRAM: _TELEGRAM_CAPS,
    CHANNEL_CLI: _PLAIN_CAPS,
    CHANNEL_API: _PLAIN_CAPS,
    CHANNEL_UNKNOWN: _EMPTY_CAPS,
}

_SESSION_CHANNEL_MAP = {
    "main": CHANNEL_SITE_CHAT,
    "docs": CHANNEL_SITE_CHAT,
    "telegram": CHANNEL_TELEGRAM,
    "cli": CHANNEL_CLI,
    "api": CHANNEL_API,
}


@dataclass
class ChannelContext:
    channel_id: str
    capabilities: dict[str, bool] = field(default_factory=dict)
    limits: dict[str, int] = field(default_factory=dict)
    tools: list[str] = field(default_factory=list)
    open_interaction_id: str | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "channel_id": self.channel_id,
            "capabilities": dict(self.capabilities),
            "limits": dict(self.limits),
            "tools": list(self.tools),
            "open_interaction_id": self.open_interaction_id,
        }


def channel_id_for_session(session_channel: str) -> str:
    return _SESSION_CHANNEL_MAP.get(str(session_channel or "").strip().lower(), CHANNEL_UNKNOWN)


def capabilities_for(channel_id: str) -> dict[str, bool]:
    return dict(_MATRICES.get(channel_id, _EMPTY_CAPS))


def limits_for(channel_id: str) -> dict[str, int]:
    limits = dict(DEFAULT_LIMITS)
    if channel_id == CHANNEL_TELEGRAM:
        limits["callback_data_bytes"] = 64
        limits["max_options"] = 16
    return limits


def interactions_enabled(
    owner_kind: str = "guest", owner_id: str = ""
) -> bool:
    from .prefs import effective_for

    return effective_for(owner_kind, owner_id)


def context_for(
    channel_id: str,
    *,
    open_interaction_id: str | None = None,
    tools_allowed: bool = True,
    owner_kind: str = "guest",
    owner_id: str = "",
) -> ChannelContext:
    caps = capabilities_for(channel_id)
    limits = limits_for(channel_id)
    tools: list[str] = []
    pref_tools = ["interactions_get", "interactions_set"]
    if owner_kind == "user":
        tools.extend(pref_tools)
    if (
        tools_allowed
        and interactions_enabled(owner_kind, owner_id)
        and channel_id in (CHANNEL_SITE_CHAT, CHANNEL_TELEGRAM, CHANNEL_CLI, CHANNEL_API)
    ):
        if open_interaction_id:
            tools = [t for t in tools if t in pref_tools] + ["ui_cancel"]
        else:
            tools = list(dict.fromkeys(tools + ["ui_prompt", "ui_cancel"]))
            if caps.get("ephemeral_status"):
                tools.append("ui_notify")
    return ChannelContext(
        channel_id=channel_id,
        capabilities=caps,
        limits=limits,
        tools=tools,
        open_interaction_id=open_interaction_id,
    )


def fragment_for(ctx: ChannelContext) -> str:
    active_caps = [name for name, on in ctx.capabilities.items() if on]
    caps_text = ",".join(active_caps) if active_caps else "none"
    tools_text = ",".join(ctx.tools) if ctx.tools else "none"
    interaction = (
        f"open:{ctx.open_interaction_id}" if ctx.open_interaction_id else "none"
    )
    return (
        f"CHANNEL: {ctx.channel_id}\n"
        f"CAPS: {caps_text}\n"
        f"LIMITS: options≤{ctx.limits.get('max_options', 32)} "
        f"fields≤{ctx.limits.get('max_fields', 24)}\n"
        f"TOOLS: {tools_text}\n"
        f"INTERACTION: {interaction}"
    )


CA_IWP_SYSTEM_FRAGMENT = (
    "# Interactive asks (CA-IWP)\n"
    "You are channel-aware. Current channel and capabilities are injected each turn as "
    "CHANNEL / CAPS / LIMITS / TOOLS / INTERACTION.\n"
    "HARD RULES when TOOLS includes ui_prompt (this is the interactive mode):\n"
    "1. You MUST call ui_prompt for every decision, confirm, choice, short form, yes/no, "
    "pick-one, multi-select, text/number/date ask, or demo of interactive widgets. Do not "
    "simulate buttons in markdown when ui_prompt is available.\n"
    "2. Do NOT fall back to numbered lists, fake [Yes]/[No] prose, or 'I would show a widget' "
    "while ui_prompt is in TOOLS. Only use degraded markdown menus when ui_prompt is absent "
    "from TOOLS.\n"
    "3. Never invent HTML, JavaScript, or custom-element tags in your visible reply. "
    "The host renders UI from tool args.\n"
    "4. Never claim buttons/widgets exist on a channel that lacks them; match wording to CHANNEL.\n"
    "5. One open interaction at a time. Wait for the ui_prompt tool result before continuing.\n"
    "6. The user may answer with the on-screen controls OR by typing a clear answer "
    "(yes/no, option label/number, free text field value, DD/MM/YYYY, or cancel). "
    "If they type a new request instead of an answer, treat the interaction as cancelled "
    "and continue with their new request.\n"
    "7. On status=cancelled or timeout, do not pretend the user agreed. Branch explicitly.\n"
    "8. Keep tool arguments compact: short labels, stable value tokens, no essays inside options.\n"
    "9. Do not use ui_prompt for long-form research answers or pure information delivery."
)