"""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 = [ { "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)