feat!: replace tool visibility flags with a single exposure option

src/bub/builtin/agent.py

@@ -481,7 +481,7 @@ async def _prepare_deferred_tools(self, tools: list[Tool], tape: Tape) -> tuple[
         )
 
         direct = [tool for tool in tools if not tool.deferred and tool.name != TOOL_DESCRIBE_TOOL_NAME]
-        deferred = {tool.name: tool for tool in tools if tool.deferred and tool.agent_use}
+        deferred = {tool.name: tool for tool in tools if tool.deferred and tool.exposure in ("auto", "direct")}
         if not deferred:
             return direct, ""
         direct.append(self.tools.get(TOOL_DESCRIBE_TOOL_NAME, tool_describe))
@@ -540,7 +540,8 @@ async def _prepare_code_mode(self, tools: list[Tool], tape: Tape) -> tuple[list[
 
         Code mode is a session setting (``state["code_mode"]``, switched by the ``code_mode``
         command) and applies only when ``run_code`` is among the allowed tools: the model then
-        sees preserved tools directly, and every other tool is callable only from code.
+        sees ``direct`` tools, and ``auto`` and ``code`` tools are callable only from code. Outside
+        code mode, ``code`` tools are dropped.
         """
         from bub.builtin.codemode import (
             CODE_MODE_STATE_KEY,
@@ -554,14 +555,14 @@ async def _prepare_code_mode(self, tools: list[Tool], tape: Tape) -> tuple[list[
         direct_tools = [tool for tool in tools if tool.name != RUN_CODE_TOOL_NAME]
         if not state.get(CODE_MODE_STATE_KEY) or len(direct_tools) == len(tools):
             state.pop(CODE_TOOLS_STATE_KEY, None)
-            return direct_tools, ""
+            return [tool for tool in direct_tools if tool.exposure != "code"], ""
 
-        code_tools = [tool for tool in direct_tools if tool.code_use]
+        code_tools = [tool for tool in direct_tools if tool.exposure in ("auto", "code")]
         state[CODE_TOOLS_STATE_KEY] = model_tools(code_tools)
         stub_path = write_tool_stub(
             code_tools, session_id=str(state.get("session_id", "")), workspace=workspace_from_state(state)
         )
-        return [tool for tool in tools if tool.preserve], render_code_mode_prompt(stub_path)
+        return [tool for tool in tools if tool.exposure == "direct"], render_code_mode_prompt(stub_path)
 
     def _system_prompt(
         self,

src/bub/builtin/codemode/__init__.py

@@ -251,7 +251,7 @@ def render_code_mode_prompt(stub_path: Path) -> str:
     )
 
 
-@tool(name=RUN_CODE_TOOL_NAME, context=True, preserve=True)
+@tool(name=RUN_CODE_TOOL_NAME, context=True, exposure="direct")
 async def run_code(code: str, timeout_seconds: int = DEFAULT_RUN_CODE_TIMEOUT_SECONDS, *, context: ToolContext) -> str:
     """Run Python code in the environment and return everything it prints.
 
@@ -299,11 +299,11 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> Any:
     return "".join(output)
 
 
-@tool(name="code_mode", context=True, agent_use=False)
+@tool(name="code_mode", context=True, exposure="command")
 async def set_code_mode(enable: bool, *, context: ToolContext) -> str:
     """Enable or disable code mode for THIS session. Invoke as the `,code_mode enable=true` command.
 
-    In code mode the model calls preserved tools directly and every other tool from
+    In code mode the model calls `direct` tools directly and every other tool from
     Python through `run_code`. Takes effect on the NEXT turn and persists across restarts.
     """
     await set_session_setting(context, CODE_MODE_STATE_KEY, enable)

src/bub/builtin/hook_impl.py

@@ -367,7 +367,7 @@ async def before_tool_call(
             code_tools = state.get(CODE_TOOLS_STATE_KEY) or ()
             available_tools = (*state["_runtime_tool_names"], *(tool_item.name for tool_item in code_tools))
         else:
-            available_tools = tuple(tool_item.name for tool_item in agent_tools)
+            available_tools = tuple(tool_item.name for tool_item in agent_tools if tool_item.exposure != "code")
         if call.tool in available_tools:
             return None
 

src/bub/builtin/spill.py

@@ -262,7 +262,7 @@ async def read(
         return SpillPage(manifest, "".join(chunks), start, stop, next_cursor, complete)
 
 
-@tool(context=True, name=SPILL_READ_TOOL_NAME, preserve=True)
+@tool(context=True, name=SPILL_READ_TOOL_NAME, exposure="direct")
 async def spill_read(
     handle: str,
     cursor: int = 0,

src/bub/builtin/tools.py

@@ -96,7 +96,7 @@ def _tool_signature(tool_item: Tool) -> str:
 
 def render_tools_prompt(tools: Iterable[Tool]) -> str:
     """Render a human-readable description of tools for builtin agent prompts."""
-    agent_tools = [tool_item for tool_item in tools if tool_item.agent_use]
+    agent_tools = [tool_item for tool_item in tools if tool_item.exposure in ("auto", "direct")]
     if not agent_tools:
         return ""
     lines = []
@@ -279,7 +279,7 @@ def _render_subagent(result: SubAgentResult) -> str:
     return result["output"] + "".join(f"[Error: {message}]" for message in result["errors"])
 
 
-@tool(context=True, preserve=True)
+@tool(context=True, exposure="direct")
 async def bash(
     command: str,
     cwd: str | None = None,
@@ -317,7 +317,7 @@ async def bash(
     return shell.output.strip() or "(no output)"
 
 
-@tool(name="bash.output", preserve=True)
+@tool(name="bash.output", exposure="direct")
 async def bash_output(shell_id: str, offset: int = 0, limit: int | None = None) -> str:
     """Read buffered output from a background shell, with optional offset/limit for incremental polling."""
     shell = shell_manager.get(shell_id)
@@ -332,14 +332,14 @@ async def bash_output(shell_id: str, offset: int = 0, limit: int | None = None)
     return f"id: {shell.shell_id}\nstatus: {shell.status}\nexit_code: {exit_code}\nnext_offset: {end}\noutput:\n{body}"
 
 
-@tool(name="bash.kill", preserve=True)
+@tool(name="bash.kill", exposure="direct")
 async def kill_bash(shell_id: str) -> str:
     """Terminate a background shell process."""
     shell = await shell_manager.terminate(shell_id)
     return f"id: {shell.shell_id}\nstatus: {shell.status}\nexit_code: {shell.returncode}"
 
 
-@tool(context=True, name="fs.read", preserve=True)
+@tool(context=True, name="fs.read", exposure="direct")
 async def fs_read(path: str, offset: int = 0, limit: int | None = None, *, context: ToolContext) -> str:
     """Read a text file and return its content. Supports optional pagination with offset and limit."""
     environment = environment_from_state(context.state)
@@ -350,7 +350,7 @@ async def fs_read(path: str, offset: int = 0, limit: int | None = None, *, conte
     return "\n".join(lines[start:end])
 
 
-@tool(context=True, name="fs.write", preserve=True)
+@tool(context=True, name="fs.write", exposure="direct")
 async def fs_write(path: str, content: str, *, context: ToolContext) -> str:
     """Write content to a text file."""
     environment = environment_from_state(context.state)
@@ -359,7 +359,7 @@ async def fs_write(path: str, content: str, *, context: ToolContext) -> str:
     return f"wrote: {resolved_path}"
 
 
-@tool(context=True, name="fs.edit", preserve=True)
+@tool(context=True, name="fs.edit", exposure="direct")
 async def fs_edit(path: str, old: str, new: str, start: int = 0, *, context: ToolContext) -> str:
     """Edit a text file by replacing old text with new text. You can specify the line number to start searching for the old text."""
     environment = environment_from_state(context.state)
@@ -398,15 +398,15 @@ def skill_describe(name: str | None = None, *, context: ToolContext) -> SkillLis
     return {"name": skill.name, "location": str(skill.location), "content": skill.body() or ""}
 
 
-@tool(context=True, name=TOOL_DESCRIBE_TOOL_NAME, preserve=True)
+@tool(context=True, name=TOOL_DESCRIBE_TOOL_NAME, exposure="direct")
 async def tool_describe(names: list[str], *, context: ToolContext) -> ToolDescriptions:
     """Load tools by name and return their definitions. Deferred tools become callable from the next step."""
     agent = _get_agent(context)
     allowed_tools = context.state.get(ALLOWED_TOOLS_STATE_KEY)
     available = {
         name: tool_item
         for name, tool_item in agent.tools.items()
-        if tool_item.agent_use and (allowed_tools is None or name in allowed_tools)
+        if tool_item.exposure in ("auto", "direct") and (allowed_tools is None or name in allowed_tools)
     }
     index = _tool_name_index(available)
     loaded = set(await loaded_tool_names(context.tape))
@@ -528,7 +528,7 @@ async def run_subagent(param: SubAgentInput, *, context: ToolContext) -> SubAgen
     return {"session_id": subagent_session, "output": output, "errors": errors}
 
 
-@tool(name="help", context=True, agent_use=False)
+@tool(name="help", context=True, exposure="command")
 def show_help(*, context: ToolContext | None = None) -> str:
     """Show a help message."""
     agent = context.state.get("_runtime_agent") if context is not None else None
@@ -554,7 +554,7 @@ def show_help(*, context: ToolContext | None = None) -> str:
     )
 
 
-@tool(name="quit", context=True, agent_use=False)
+@tool(name="quit", context=True, exposure="command")
 async def quit_tool(*, context: ToolContext) -> str:
     """Abort the tasks of the current session. DO NOT use it in a normal workflow."""
     agent = _get_agent(context)
@@ -564,7 +564,7 @@ async def quit_tool(*, context: ToolContext) -> str:
     return "Session tasks stopped."
 
 
-@tool(name="model", context=True, agent_use=False)
+@tool(name="model", context=True, exposure="command")
 async def set_model(model_id: str, *, context: ToolContext) -> str:
     """Switch the model for THIS session. Invoke as the `,model <model_id>` command.
 
@@ -577,7 +577,7 @@ async def set_model(model_id: str, *, context: ToolContext) -> str:
     return f"Session model set to {model_id} (applies from the next turn)."
 
 
-@tool(name="reasoning_effort", context=True, agent_use=False)
+@tool(name="reasoning_effort", context=True, exposure="command")
 async def set_reasoning_effort(reasoning_effort: str, *, context: ToolContext) -> str:
     """Set the reasoning effort for this session starting from the next turn."""
     reasoning_effort = reasoning_effort.strip()

src/bub/tools.py

@@ -8,7 +8,7 @@
 import time
 from collections.abc import Awaitable, Callable, Iterable, Sequence
 from dataclasses import dataclass, field, replace
-from typing import TYPE_CHECKING, Any, Protocol, get_type_hints, overload
+from typing import TYPE_CHECKING, Any, Literal, Protocol, get_args, get_type_hints, overload
 
 from loguru import logger
 from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError, validate_call
@@ -134,6 +134,16 @@ def validate_target(*args: Any, **kwargs: Any) -> tuple[tuple[Any, ...], dict[st
     return validate_call(validate_target)
 
 
+ToolExposure = Literal["auto", "direct", "code", "command"]
+"""Where a tool can be called from.
+
+- ``auto``: by the model directly, or only from ``run_code`` in code mode.
+- ``direct``: by the model directly, also in code mode; never from ``run_code``.
+- ``code``: only from ``run_code`` in code mode; never by the model directly.
+- ``command``: only as a comma command; never by the model or from code.
+"""
+
+
 @dataclass(frozen=True)
 class Tool:
     """A callable unit the model can invoke."""
@@ -143,22 +153,17 @@ class Tool:
     description: str = ""
     parameters: dict[str, Any] = field(default_factory=dict)
     context: bool = False
-    agent_use: bool = True
+    exposure: ToolExposure = "auto"
+    """Where the tool can be called from, see ``ToolExposure``."""
     renderer: Callable[[Any], str] | None = None
-    preserve: bool = False
-    """Keep the tool directly callable by the model in code mode; others are reachable only from code."""
     output_schema: dict[str, Any] | None = None
     """JSON schema of the structured result, used to describe the tool to model-written code."""
     deferred: bool = False
     """Load the tool on demand: only its name is listed until ``tool.describe`` loads its definition."""
 
-    @property
-    def code_use(self) -> bool:
-        """Whether the tool is callable from model-written code (``tools.*`` in ``run_code``).
-
-        Preserved tools stay model-facing only, and tools hidden from the agent are never exposed to code.
-        """
-        return self.agent_use and not self.preserve
+    def __post_init__(self) -> None:
+        if self.exposure not in get_args(ToolExposure):
+            raise ValueError(f"Tool '{self.name}' has unknown exposure {self.exposure!r}.")
 
     def run(self, *args: Any, **kwargs: Any) -> Any:
         return self.handler(*args, **kwargs)
@@ -188,9 +193,8 @@ def from_callable(
         name: str | None = None,
         description: str | None = None,
         context: bool = False,
-        agent_use: bool = True,
+        exposure: ToolExposure = "auto",
         renderer: Callable[[Any], str] | None = None,
-        preserve: bool = False,
         deferred: bool = False,
     ) -> Tool:
         signature = inspect.signature(func)
@@ -215,17 +219,20 @@ def validated(*args: Any, **kwargs: Any) -> Any:
             parameters=parameters,
             handler=validated,
             context=context,
-            agent_use=agent_use,
+            exposure=exposure,
             renderer=renderer,
-            preserve=preserve,
             output_schema=_output_schema(func),
             deferred=deferred,
         )
 
 
 def model_tools(tools: Iterable[Tool]) -> list[Tool]:
-    """Convert agent-enabled runtime tools into model-safe aliases."""
-    return [replace(tool_item, name=tool_item.name.replace(".", "_")) for tool_item in tools if tool_item.agent_use]
+    """Convert tools callable by the model or code into model-safe aliases; comma commands are dropped."""
+    return [
+        replace(tool_item, name=tool_item.name.replace(".", "_"))
+        for tool_item in tools
+        if tool_item.exposure != "command"
+    ]
 
 
 @dataclass(frozen=True)
@@ -559,9 +566,8 @@ def tool(
     model: type[BaseModel] | None = ...,
     description: str | None = ...,
     context: bool = ...,
-    agent_use: bool = ...,
+    exposure: ToolExposure = ...,
     renderer: Callable[[Any], str] | None = ...,
-    preserve: bool = ...,
     deferred: bool = ...,
 ) -> Tool: ...
 
@@ -574,9 +580,8 @@ def tool(
     model: type[BaseModel] | None = ...,
     description: str | None = ...,
     context: bool = ...,
-    agent_use: bool = ...,
+    exposure: ToolExposure = ...,
     renderer: Callable[[Any], str] | None = ...,
-    preserve: bool = ...,
     deferred: bool = ...,
 ) -> Callable[[Callable], Tool]: ...
 
@@ -588,15 +593,15 @@ def tool(
     model: type[BaseModel] | None = None,
     description: str | None = None,
     context: bool = False,
-    agent_use: bool = True,
+    exposure: ToolExposure = "auto",
     renderer: Callable[[Any], str] | None = None,
-    preserve: bool = False,
     deferred: bool = False,
 ) -> Tool | Callable[[Callable], Tool]:
     """Decorator to convert a function into a Tool instance.
 
     Tools should return structured results; ``renderer`` turns such a result into the
     plain text shown to the model outside code mode (defaults to JSON for non-strings).
+    ``exposure`` decides where the tool can be called from (see ``ToolExposure``), and
     ``deferred`` tools are loaded on demand through ``tool.describe``.
     """
 
@@ -618,9 +623,8 @@ def handler(*args: Any, **kwargs: Any) -> Any:
                 parameters=model.model_json_schema(),
                 handler=handler,
                 context=context,
-                agent_use=agent_use,
+                exposure=exposure,
                 renderer=renderer,
-                preserve=preserve,
                 output_schema=_output_schema(func),
                 deferred=deferred,
             )
@@ -630,9 +634,8 @@ def handler(*args: Any, **kwargs: Any) -> Any:
                 name=name,
                 description=description,
                 context=context,
-                agent_use=agent_use,
+                exposure=exposure,
                 renderer=renderer,
-                preserve=preserve,
                 deferred=deferred,
             )
         tool_instance = _add_logging(result)

tests/test_builtin_agent.py

@@ -411,7 +411,7 @@ def denied_agent_tool() -> str:
 
 
 @pytest.mark.asyncio
-async def test_agent_run_excludes_tools_disabled_for_agent_use() -> None:
+async def test_agent_run_excludes_command_tools() -> None:
     visible_name = "tests.visible_agent_tool"
     internal_name = "tests.internal_agent_tool"
     REGISTRY.pop(visible_name, None)
@@ -421,7 +421,7 @@ async def test_agent_run_excludes_tools_disabled_for_agent_use() -> None:
     def visible_agent_tool() -> str:
         return "visible"
 
-    @tool(name=internal_name, description="Internal tool", agent_use=False)
+    @tool(name=internal_name, description="Internal tool", exposure="command")
     def internal_agent_tool() -> str:
         return "internal"
 

tests/test_builtin_tools.py

@@ -103,8 +103,8 @@ def test_render_tools_prompt_returns_empty_string_for_empty_input() -> None:
     assert render_tools_prompt([]) == ""
 
 
-def test_render_tools_prompt_excludes_tools_disabled_for_agent_use() -> None:
-    internal_tool = Tool(name="tests.internal", handler=lambda: None, agent_use=False)
+def test_render_tools_prompt_excludes_command_tools() -> None:
+    internal_tool = Tool(name="tests.internal", handler=lambda: None, exposure="command")
 
     assert render_tools_prompt([internal_tool]) == ""
 
@@ -188,7 +188,7 @@ async def test_set_model_overwrites_previous_model(tmp_path) -> None:
 def test_set_reasoning_effort_is_registered_for_internal_use() -> None:
     assert REGISTRY["reasoning_effort"] is set_reasoning_effort
     assert set_reasoning_effort.context is True
-    assert set_reasoning_effort.agent_use is False
+    assert set_reasoning_effort.exposure == "command"
     assert set_reasoning_effort.parameters == {
         "type": "object",
         "properties": {"reasoning_effort": {"type": "string"}},