tools.py

"""Tool surface: what the model calls. Thin wrapper over an Orchestrator,
translating tool calls to model-facing strings.

Only ONE tool: validate_answer -- a WRITE that judges + mutates state and must
not be self-judged. The current quest is a READ of state we already track, so it
is NOT here: Orchestrator.quest_view() supplies it and we inject it into the
prompt (narrator.present_quest) -- no tool round-trip for data we own.
"""
from __future__ import annotations

import json

from config import MAX_ATTEMPT
from core.game_state import GameState
from core.orchestrator import Orchestrator
from core.quest_bank import QuestBank
from core.validation import Validator

# Model-facing tool schemas (what the LLM sees), in Qwen2.5's trained
# (Hermes/OpenAI) nested shape: {"type": "function", "function": {...}}.
# The chat template renders these into <tools>...</tools>.
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "validate_answer",
            "description": (
                "Validate the player's answer to the current question. Returns a "
                "verdict (correct/partial/wrong/lost) and maybe a hint."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "the player's answer"}
                },
                "required": ["text"],
            },
        },
    },
]


class GameSession:
    # whitelist (derived from TOOLS): the only methods the model may dispatch to
    TOOL_NAMES = {t["function"]["name"] for t in TOOLS}

    def __init__(self, bank: QuestBank | None = None, validator: Validator | None = None):
        self.orchestrator = Orchestrator(bank, validator=validator)

    @property
    def state(self) -> GameState:
        return self.orchestrator.state

    def validate_answer(self, text: str) -> str:
        """Model-facing: commit an answer; returns a short string for the model.

        The strike count rides here (this is the action that changes it), so the
        model always sees the current count from the tool channel.
        """
        res = self.orchestrator.submit(text)
        if res.verdict in ("wrong", "partial"):
            return f"{res.verdict} (strike {self.state.attempt}/{MAX_ATTEMPT}) -- hint: {res.hint}"
        if res.verdict == "lost":
            return f"lost -- {res.hint}"          # hint is the revealed answer
        return res.verdict                        # correct

    def dispatch(self, name: str, arguments: dict) -> str:
        """Route a parsed tool call to a whitelisted method; return a string for the model.

        Whitelisted to the declared TOOLS so the model can't reach arbitrary methods.
        """
        if name not in self.TOOL_NAMES:
            return f"error: unknown tool '{name}'"
        try:
            result = getattr(self, name)(**arguments)
        except Exception as exc:
            return f"error: {name} failed -- {exc}"      # model sees it, can retry
        return result if isinstance(result, str) else json.dumps(result)