ai-video-fullstack/backend/services/system_tools.py

"""Contracts and adapters for platform-owned built-in conversation tools."""

from __future__ import annotations

from typing import Any, Mapping

from models import RuntimeTool
from services.state_expressions import validate_update_rules


SYSTEM_TOOL_SPECS: dict[str, dict[str, str]] = {
    "end_conversation": {
        "name": "结束对话",
        "function_name": "end_conversation",
        "description": "礼貌地结束本次对话。",
    },
    "update_state": {
        "name": "更新状态",
        "function_name": "update_state",
        "description": "静默更新已经声明并明确授权的会话状态。",
    },
    "skip_turn": {
        "name": "跳过本轮",
        "function_name": "skip_turn",
        "description": "跳过当前轮次，不生成语音回复。",
    },
    "request_human_handoff": {
        "name": "转接人工",
        "function_name": "request_human_handoff",
        "description": "提交人工接管请求，并继续告知用户当前进展。",
    },
}

SYSTEM_TOOL_KINDS = frozenset(SYSTEM_TOOL_SPECS)
RESERVED_TOOL_NAMES = frozenset(
    spec["function_name"] for spec in SYSTEM_TOOL_SPECS.values()
)


def system_tool_kind(definition: Mapping[str, Any] | None) -> str | None:
    """Compatibility helper for old fixtures and the one-time migration."""

    config = (definition or {}).get("config")
    if not isinstance(config, Mapping):
        return None
    kind = str(config.get("kind") or "")
    return kind if kind in SYSTEM_TOOL_KINDS else None


def runtime_system_tools(
    built_in_tools: Mapping[str, Any] | None,
) -> list[RuntimeTool]:
    """Adapt immutable built-in configs to existing brain tool builders."""

    result: list[RuntimeTool] = []
    aliases = {
        "endConversation": "end_conversation",
        "updateState": "update_state",
        "skipTurn": "skip_turn",
        "requestHumanHandoff": "request_human_handoff",
    }
    config_aliases = {
        "messageType": "message_type",
        "customMessage": "custom_message",
        "captureReason": "capture_reason",
        "allowEndInterruption": "allow_end_interruption",
    }
    for raw_kind, config in (built_in_tools or {}).items():
        kind = aliases.get(str(raw_kind), str(raw_kind))
        if kind not in SYSTEM_TOOL_SPECS or not isinstance(config, Mapping):
            continue
        spec = SYSTEM_TOOL_SPECS[kind]
        normalized_config = dict(config)
        for camel, snake in config_aliases.items():
            if camel in normalized_config and snake not in normalized_config:
                normalized_config[snake] = normalized_config.pop(camel)
        result.append(
            RuntimeTool(
                id=f"builtin_{kind}",
                name=spec["name"],
                function_name=spec["function_name"],
                type="system",
                description=str(
                    normalized_config.get("description") or spec["description"]
                ),
                definition={
                    "type": "system",
                    "config": {"kind": kind, **normalized_config},
                },
            )
        )
    return result


def state_update_properties(
    definitions: Mapping[str, Mapping[str, Any]] | None,
    *,
    rules: list[Mapping[str, Any]] | None = None,
) -> dict[str, dict[str, Any]]:
    """Build the LLM schema from explicit v4 expression rules."""

    return validate_update_rules(rules or [], definitions, allow_llm=True)