kiln_ai.tool_testing

 36class FakeToolBridge:
 37    """Registry-backed stand-in for the sandbox tool-call bridge.
 38
 39    Authors drive it through the ``kiln_tools`` fixture:
 40
 41    * :meth:`set` registers a reply (static string or callable) for a tool name.
 42    * :meth:`set_error` registers an exception a tool name should raise.
 43    * :attr:`calls` records every call in order for assertions.
 44    * :meth:`list_tools` returns the registered declarations.
 45
 46    An unregistered name raises :class:`ToolNotAllowed`, matching the runtime's
 47    allowlist miss.
 48    """
 49
 50    def __init__(self) -> None:
 51        self._lock = threading.Lock()
 52        self._replies: dict[str, ToolReply] = {}
 53        self._errors: dict[str, BaseException] = {}
 54        self._declarations: dict[str, dict[str, Any]] = {}
 55        self.calls: list[RecordedToolCall] = []
 56
 57    # -- registration --
 58
 59    def set(
 60        self,
 61        name: str,
 62        reply: ToolReply,
 63        *,
 64        declaration: dict[str, Any] | None = None,
 65    ) -> None:
 66        """Register *reply* for tool *name*.
 67
 68        *reply* is a ``str`` (returned verbatim, matching the string-returns
 69        contract) or a callable ``(**kwargs) -> str``. Optionally attach a
 70        *declaration* dict surfaced by :meth:`list_tools`.
 71        """
 72        if not isinstance(reply, str) and not callable(reply):
 73            raise TypeError(
 74                f"kiln_tools.set('{name}', ...): reply must be a str or a "
 75                f"callable (**kwargs) -> str, got {type(reply).__name__}"
 76            )
 77        with self._lock:
 78            self._replies[name] = reply
 79            self._errors.pop(name, None)
 80            self._declarations[name] = self._build_declaration(name, declaration)
 81
 82    def set_error(
 83        self,
 84        name: str,
 85        exc: BaseException,
 86        *,
 87        declaration: dict[str, Any] | None = None,
 88    ) -> None:
 89        """Register *exc* to be raised whenever tool *name* is called."""
 90        if not isinstance(exc, BaseException):
 91            raise TypeError(
 92                f"kiln_tools.set_error('{name}', ...): exc must be an exception "
 93                f"instance, got {type(exc).__name__}"
 94            )
 95        with self._lock:
 96            self._errors[name] = exc
 97            self._replies.pop(name, None)
 98            self._declarations[name] = self._build_declaration(name, declaration)
 99
100    def reset(self) -> None:
101        """Clear all registrations and the call log (per-test reset)."""
102        with self._lock:
103            self._replies.clear()
104            self._errors.clear()
105            self._declarations.clear()
106            self.calls.clear()
107
108    # -- ToolBridge protocol --
109
110    def call(self, name: str, args: tuple[Any, ...], kwargs: dict[str, Any]) -> str:
111        """Resolve a tool call against the registry.
112
113        Records the call first, then dispatches in the same order the runtime
114        does (``tools/code_tool.py``): the not-allowed check comes FIRST, so an
115        unregistered name raises :class:`ToolNotAllowed` regardless of arguments;
116        only for a registered name do positional arguments raise
117        :class:`ToolCallError` (tools are keyword-only); then a registered error
118        is raised, otherwise the static or callable reply is returned.
119        """
120        with self._lock:
121            self.calls.append(RecordedToolCall(name=name, arguments=dict(kwargs)))
122            error = self._errors.get(name)
123            has_reply = name in self._replies
124            reply = self._replies.get(name)
125
126        # A name is "registered" if it has either a reply or an error. Mirror the
127        # runtime's not-allowed branch, which precedes the positional-args branch.
128        if not has_reply and error is None:
129            raise ToolNotAllowed(
130                tool=name,
131                message=(
132                    f"Tool '{name}' is not registered with kiln_tools. Register a "
133                    f"reply with kiln_tools.set('{name}', ...) or an error with "
134                    f"kiln_tools.set_error('{name}', ...)."
135                ),
136            )
137
138        if args:
139            raise ToolCallError(
140                tool=name,
141                message=(
142                    f"Tool '{name}' was called with positional arguments; tool "
143                    f"calls must use keyword arguments."
144                ),
145            )
146
147        if error is not None:
148            raise error
149
150        if callable(reply):
151            result = reply(**kwargs)
152            if not isinstance(result, str):
153                raise TypeError(
154                    f"kiln_tools reply callable for '{name}' must return a str, "
155                    f"got {type(result).__name__}"
156                )
157            return result
158
159        # A non-callable reply is a str (enforced at set()).
160        assert isinstance(reply, str)
161        return reply
162
163    def list_tools(self) -> list[dict[str, Any]]:
164        """Return the registered tool declarations (name + supplied fields)."""
165        with self._lock:
166            return [dict(decl) for decl in self._declarations.values()]
167
168    # -- internals --
169
170    @staticmethod
171    def _build_declaration(
172        name: str, declaration: dict[str, Any] | None
173    ) -> dict[str, Any]:
174        decl: dict[str, Any] = dict(declaration) if declaration else {}
175        decl["name"] = name
176        return decl

28@dataclass
29class RecordedToolCall:
30    """One recorded tool call: the tool ``name`` and its keyword ``arguments``."""
31
32    name: str
33    arguments: dict[str, Any] = field(default_factory=dict)