From 85a452e5c760b91be55bb4af6932e9ce7e487a4c Mon Sep 17 00:00:00 2001 From: chengyongru <61816729+chengyongru@users.noreply.github.com> Date: Mon, 10 Aug 2026 16:44:26 +0800 Subject: [PATCH] refactor(agent): replace reflective runtime state access (#5319) --- nanobot/agent/loop.py | 14 +- nanobot/agent/subagent.py | 5 + nanobot/agent/tools/loader.py | 2 +- nanobot/agent/tools/runtime_control.py | 319 ++++++++++++++ nanobot/agent/tools/runtime_state.py | 76 ---- nanobot/agent/tools/self.py | 395 ++++++++++-------- tests/agent/test_self_model_preset.py | 24 +- tests/agent/tools/test_runtime_control.py | 225 ++++++++++ tests/agent/tools/test_self_tool.py | 143 ++++--- .../tools/test_self_tool_runtime_sync.py | 26 +- 10 files changed, 877 insertions(+), 352 deletions(-) create mode 100644 nanobot/agent/tools/runtime_control.py delete mode 100644 nanobot/agent/tools/runtime_state.py create mode 100644 tests/agent/tools/test_runtime_control.py diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index 46a33b53d..9fa12e003 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -36,6 +36,7 @@ from nanobot.agent.tools.exec_session import ExecSessionManager from nanobot.agent.tools.file_state import FileStateStore, bind_file_states, reset_file_states from nanobot.agent.tools.message import MessageTool from nanobot.agent.tools.registry import ToolRegistry +from nanobot.agent.tools.runtime_control import AgentRuntimeControl from nanobot.agent.tools.self import MyTool from nanobot.agent.turn_delivery import ( TurnDelivery, @@ -197,6 +198,11 @@ class AgentLoop: def tool_names(self) -> list[str]: return self.tools.tool_names + @property + def last_usage(self) -> Mapping[str, int]: + """Latest aggregate usage exposed through the runtime-control snapshot.""" + return self._last_usage + @property def provider(self) -> LLMProvider: """Provider selected for future turn admissions.""" @@ -448,7 +454,6 @@ class AgentLoop: if model_preset: self.set_model_preset(model_preset, publish_update=False) self._register_default_tools(provider_snapshot_loader=provider_snapshot_loader) - self._runtime_vars: dict[str, Any] = {} self._current_iteration: int = 0 self.commands = CommandRouter() register_builtin_commands(self.commands) @@ -623,10 +628,13 @@ class AgentLoop: loader = ToolLoader() registered = loader.load(ctx, self.tools) - # MyTool needs runtime state reference — manual registration + # MyTool receives only the explicit runtime-control capability. if self.tools_config.my.enable: self.tools.register( - MyTool(runtime_state=self, modify_allowed=self.tools_config.my.allow_set) + MyTool( + runtime_control=AgentRuntimeControl(self), + modify_allowed=self.tools_config.my.allow_set, + ) ) registered.append("my") diff --git a/nanobot/agent/subagent.py b/nanobot/agent/subagent.py index c4d151f74..e46146ea1 100644 --- a/nanobot/agent/subagent.py +++ b/nanobot/agent/subagent.py @@ -5,6 +5,7 @@ import json import time import uuid import warnings +from collections.abc import Mapping from dataclasses import dataclass, field from pathlib import Path from typing import Any, Callable, TypedDict @@ -157,6 +158,10 @@ class SubagentManager: self._task_statuses: dict[str, SubagentStatus] = {} self._session_tasks: dict[str, set[str]] = {} # session_key -> {task_id, ...} + def runtime_statuses(self) -> Mapping[str, SubagentStatus]: + """Return the observable task statuses used by runtime-control snapshots.""" + return self._task_statuses + def set_provider(self, provider: LLMProvider, model: str) -> None: """Update the deprecated runtime source used by legacy ``spawn`` calls.""" warnings.warn( diff --git a/nanobot/agent/tools/loader.py b/nanobot/agent/tools/loader.py index 27760cec6..85cc8283d 100644 --- a/nanobot/agent/tools/loader.py +++ b/nanobot/agent/tools/loader.py @@ -19,7 +19,7 @@ if TYPE_CHECKING: _SKIP_MODULES = frozenset({ "base", "schema", "registry", "context", "loader", "config", - "file_state", "sandbox", "mcp", "__init__", "runtime_state", + "file_state", "sandbox", "mcp", "__init__", "runtime_control", }) diff --git a/nanobot/agent/tools/runtime_control.py b/nanobot/agent/tools/runtime_control.py new file mode 100644 index 000000000..21fae0c70 --- /dev/null +++ b/nanobot/agent/tools/runtime_control.py @@ -0,0 +1,319 @@ +"""Explicit runtime state boundary used by :class:`MyTool`.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Protocol, TypeAlias, runtime_checkable + +if TYPE_CHECKING: + from nanobot.agent.subagent import SubagentManager, SubagentStatus + from nanobot.agent.tools.shell import ExecToolConfig + from nanobot.agent.tools.web import WebToolsConfig + from nanobot.config.schema import ModelPresetConfig + from nanobot.utils.llm_runtime import LLMRuntime + + +JsonScalar: TypeAlias = str | int | float | bool | None +JsonValue: TypeAlias = JsonScalar | list["JsonValue"] | dict[str, "JsonValue"] + + +RUNTIME_SNAPSHOT_KEYS = frozenset({ + "model", + "model_preset", + "model_presets", + "max_iterations", + "context_window_tokens", + "workspace", + "provider_retry_mode", + "max_tool_result_chars", + "current_iteration", + "_current_iteration", + "tool_names", + "web_config", + "exec_config", + "subagents", + "_last_usage", +}) + +RUNTIME_COMMAND_KEYS = frozenset({ + "model", + "model_preset", + "max_iterations", + "context_window_tokens", + "provider_retry_mode", + "max_tool_result_chars", + "workspace", +}) + + +@dataclass(frozen=True, slots=True) +class RuntimeSnapshot: + """Detached, allowlisted values available to self-inspection.""" + + model: str + model_preset: str | None + model_presets: dict[str, dict[str, object]] + max_iterations: int + context_window_tokens: int + workspace: Path | str + provider_retry_mode: str + max_tool_result_chars: int + current_iteration: int + tool_names: list[str] + web_config: dict[str, object] + exec_config: dict[str, object] + subagent_statuses: dict[str, dict[str, object]] + last_usage: dict[str, int] + scratchpad: dict[str, JsonValue] + + def as_mapping(self) -> Mapping[str, object]: + """Return the fixed public names understood by ``MyTool``.""" + values: dict[str, object] = { + "model": self.model, + "model_preset": self.model_preset, + "model_presets": self.model_presets, + "max_iterations": self.max_iterations, + "context_window_tokens": self.context_window_tokens, + "workspace": self.workspace, + "provider_retry_mode": self.provider_retry_mode, + "max_tool_result_chars": self.max_tool_result_chars, + "current_iteration": self.current_iteration, + "_current_iteration": self.current_iteration, + "tool_names": self.tool_names, + "web_config": self.web_config, + "exec_config": self.exec_config, + "subagents": {"_task_statuses": self.subagent_statuses}, + "_last_usage": self.last_usage, + } + assert values.keys() == RUNTIME_SNAPSHOT_KEYS + return values + + +@runtime_checkable +class RuntimeControl(Protocol): + """The complete runtime capability exposed to ``MyTool``.""" + + def snapshot(self) -> RuntimeSnapshot: ... + + def set_model(self, model: str) -> LLMRuntime: ... + + def set_model_preset( + self, + name: str, + *, + session_key: str | None, + ) -> LLMRuntime: ... + + def set_max_iterations(self, value: int) -> None: ... + + def set_context_window_tokens(self, value: int) -> LLMRuntime: ... + + def set_provider_retry_mode(self, value: str) -> None: ... + + def set_max_tool_result_chars(self, value: int) -> None: ... + + def set_workspace_display(self, value: str) -> None: ... + + def set_scratchpad(self, key: str, value: JsonValue, *, max_keys: int) -> None: ... + + +class _RuntimeControlTarget(Protocol): + """Narrow structural dependency required by ``AgentRuntimeControl``.""" + + max_iterations: int + provider_retry_mode: str + max_tool_result_chars: int + web_config: WebToolsConfig + exec_config: ExecToolConfig + subagents: SubagentManager + + @property + def model(self) -> str: ... + + @property + def model_preset(self) -> str | None: ... + + @property + def model_presets(self) -> Mapping[str, ModelPresetConfig]: ... + + @property + def context_window_tokens(self) -> int: ... + + @property + def workspace(self) -> Path: ... + + @property + def current_iteration(self) -> int: ... + + @property + def tool_names(self) -> list[str]: ... + + @property + def last_usage(self) -> Mapping[str, int]: ... + + def set_runtime_model(self, model: str) -> LLMRuntime: ... + + def set_runtime_context_window(self, context_window_tokens: int) -> LLMRuntime: ... + + def set_model_preset(self, name: str | None) -> LLMRuntime: ... + + def set_session_model_preset(self, session_key: str, name: str) -> LLMRuntime: ... + + +class AgentRuntimeControl: + """Allowlisted adapter from agent-loop state to ``RuntimeControl``.""" + + def __init__(self, target: _RuntimeControlTarget) -> None: + self.__target = target + self.__scratchpad: dict[str, JsonValue] = {} + self.__workspace_display: str | None = None + + def snapshot(self) -> RuntimeSnapshot: + target = self.__target + return RuntimeSnapshot( + model=target.model, + model_preset=target.model_preset, + model_presets=_snapshot_model_presets(target.model_presets), + max_iterations=target.max_iterations, + context_window_tokens=target.context_window_tokens, + workspace=( + self.__workspace_display + if self.__workspace_display is not None + else target.workspace + ), + provider_retry_mode=target.provider_retry_mode, + max_tool_result_chars=target.max_tool_result_chars, + current_iteration=target.current_iteration, + tool_names=list(target.tool_names), + web_config=_snapshot_web_config(target.web_config), + exec_config=_snapshot_exec_config(target.exec_config), + subagent_statuses=_snapshot_subagent_statuses(target.subagents), + last_usage=dict(target.last_usage), + scratchpad=_snapshot_json_mapping(self.__scratchpad), + ) + + def set_model(self, model: str) -> LLMRuntime: + return self.__target.set_runtime_model(model) + + def set_model_preset( + self, + name: str, + *, + session_key: str | None, + ) -> LLMRuntime: + if session_key is not None: + return self.__target.set_session_model_preset(session_key, name) + return self.__target.set_model_preset(name) + + def set_max_iterations(self, value: int) -> None: + self.__target.max_iterations = value + self.__target.subagents.max_iterations = value + + def set_context_window_tokens(self, value: int) -> LLMRuntime: + return self.__target.set_runtime_context_window(value) + + def set_provider_retry_mode(self, value: str) -> None: + self.__target.provider_retry_mode = value + + def set_max_tool_result_chars(self, value: int) -> None: + self.__target.max_tool_result_chars = value + + def set_workspace_display(self, value: str) -> None: + """Preserve MyTool display compatibility without changing path enforcement.""" + self.__workspace_display = value + + def set_scratchpad(self, key: str, value: JsonValue, *, max_keys: int) -> None: + if key not in self.__scratchpad and len(self.__scratchpad) >= max_keys: + raise ValueError(f"scratchpad is full (max {max_keys} keys)") + self.__scratchpad[key] = value + + +def _snapshot_model_presets( + presets: Mapping[str, ModelPresetConfig], +) -> dict[str, dict[str, object]]: + return { + name: { + "label": preset.label, + "model": preset.model, + "provider": preset.provider, + "max_tokens": preset.max_tokens, + "context_window_tokens": preset.context_window_tokens, + "temperature": preset.temperature, + "reasoning_effort": preset.reasoning_effort, + } + for name, preset in presets.items() + } + + +def _snapshot_web_config(config: WebToolsConfig) -> dict[str, object]: + return { + "enable": config.enable, + # Proxy URLs may embed credentials. Presence is enough for diagnosis. + "proxy": "" if config.proxy else config.proxy, + "user_agent": config.user_agent, + "search": { + "provider": config.search.provider, + "base_url": config.search.base_url, + "max_results": config.search.max_results, + "timeout": config.search.timeout, + }, + "fetch": { + "use_jina_reader": config.fetch.use_jina_reader, + }, + } + + +def _snapshot_exec_config(config: ExecToolConfig) -> dict[str, object]: + return { + "enable": config.enable, + "timeout": config.timeout, + "path_prepend": config.path_prepend, + "path_append": config.path_append, + "sandbox": config.sandbox, + "sandbox_ro_binds": list(config.sandbox_ro_binds), + "sandbox_rw_binds": list(config.sandbox_rw_binds), + "allowed_env_keys": list(config.allowed_env_keys), + "allow_patterns": list(config.allow_patterns), + "deny_patterns": list(config.deny_patterns), + } + + +def _snapshot_subagent_statuses( + manager: SubagentManager, +) -> dict[str, dict[str, object]]: + return { + task_id: _snapshot_subagent_status(status) + for task_id, status in manager.runtime_statuses().items() + } + + +def _snapshot_subagent_status(status: SubagentStatus) -> dict[str, object]: + return { + "task_id": status.task_id, + "label": status.label, + "task_description": status.task_description, + "started_at": status.started_at, + "phase": status.phase, + "iteration": status.iteration, + "tool_events": [dict(event) for event in status.tool_events], + "usage": dict(status.usage), + "stop_reason": status.stop_reason, + "error": status.error, + } + + +def _snapshot_json_mapping(values: Mapping[str, JsonValue]) -> dict[str, JsonValue]: + return {key: _snapshot_json_value(value) for key, value in values.items()} + + +def _snapshot_json_value(value: JsonValue) -> JsonValue: + if isinstance(value, list): + return [_snapshot_json_value(item) for item in value] + if isinstance(value, dict): + return { + key: _snapshot_json_value(item) + for key, item in value.items() + } + return value diff --git a/nanobot/agent/tools/runtime_state.py b/nanobot/agent/tools/runtime_state.py deleted file mode 100644 index 3efe8e870..000000000 --- a/nanobot/agent/tools/runtime_state.py +++ /dev/null @@ -1,76 +0,0 @@ -"""RuntimeState protocol: agent loop state exposed to MyTool.""" - -from __future__ import annotations - -from pathlib import Path -from typing import TYPE_CHECKING, Any, Protocol - -if TYPE_CHECKING: - from nanobot.agent.subagent import SubagentManager - from nanobot.agent.tools.shell import ExecToolConfig - from nanobot.agent.tools.web import WebToolsConfig - from nanobot.utils.llm_runtime import LLMRuntime - - -class RuntimeState(Protocol): - """Minimum contract that MyTool requires from its runtime state provider. - - In practice, this is always satisfied by ``AgentLoop``. MyTool also - accesses arbitrary attributes dynamically (via ``getattr`` / ``setattr``) - for dot-path inspection and modification; those paths are validated at - runtime rather than by this protocol. - """ - - @property - def model(self) -> str: ... - - @property - def max_iterations(self) -> int: ... - - @property - def current_iteration(self) -> int: ... - - @property - def tool_names(self) -> list[str]: ... - - @property - def workspace(self) -> Path: ... - - @property - def provider_retry_mode(self) -> str: ... - - @property - def max_tool_result_chars(self) -> int: ... - - @property - def context_window_tokens(self) -> int: ... - - @property - def web_config(self) -> WebToolsConfig: ... - - @property - def exec_config(self) -> ExecToolConfig: ... - - @property - def subagents(self) -> SubagentManager: ... - - @property - def _runtime_vars(self) -> dict[str, Any]: ... - - @property - def _last_usage(self) -> dict[str, int]: ... - - def _sync_subagent_runtime_limits(self) -> None: ... - - def set_runtime_model(self, model: str) -> LLMRuntime: ... - - def set_runtime_context_window(self, context_window_tokens: int) -> LLMRuntime: ... - - def set_session_model_preset( - self, - session_key: str, - name: str, - ) -> LLMRuntime: ... - - @property - def model_preset(self) -> str | None: ... diff --git a/nanobot/agent/tools/self.py b/nanobot/agent/tools/self.py index 3cad85242..075c43806 100644 --- a/nanobot/agent/tools/self.py +++ b/nanobot/agent/tools/self.py @@ -1,8 +1,7 @@ """MyTool: runtime state inspection and configuration for the agent loop.""" -# RuntimeState intentionally exposes a narrow set of AgentLoop internals to -# this manually registered tool. Tool.execute accepts heterogeneous schemas. -# pyright: reportPrivateUsage=false, reportIncompatibleMethodOverride=false +# Tool.execute accepts heterogeneous schemas. +# pyright: reportIncompatibleMethodOverride=false from __future__ import annotations @@ -14,7 +13,13 @@ from loguru import logger from nanobot.agent.tools.base import Tool, ToolResult from nanobot.agent.tools.context import current_request_context, current_request_session_key -from nanobot.agent.tools.runtime_state import RuntimeState +from nanobot.agent.tools.runtime_control import ( + RUNTIME_COMMAND_KEYS, + RUNTIME_SNAPSHOT_KEYS, + JsonValue, + RuntimeControl, + RuntimeSnapshot, +) from nanobot.config_base import Base if TYPE_CHECKING: @@ -28,25 +33,28 @@ class MyToolConfig(Base): allow_set: bool = False -def _has_real_attr(obj: Any, key: str) -> bool: - """Check if obj has a real (explicitly set) attribute, not auto-generated by mock.""" - if isinstance(obj, dict): - return key in obj - d = getattr(obj, "__dict__", None) - if d is not None and key in d: - return True - for cls in type(obj).__mro__: - if key in cls.__dict__: - return True - return False - - def _is_subagent_status(value: object) -> TypeGuard[SubagentStatus]: from nanobot.agent.subagent import SubagentStatus return isinstance(value, SubagentStatus) +def _is_subagent_status_snapshot(value: object) -> TypeGuard[Mapping[str, object]]: + if not isinstance(value, Mapping): + return False + return all( + field in value + for field in ("task_id", "label", "task_description", "started_at", "phase") + ) + + +def _is_string_mapping(value: object) -> TypeGuard[Mapping[str, object]]: + if not isinstance(value, Mapping): + return False + mapping = cast(Mapping[object, object], value) + return all(isinstance(key, str) for key in mapping) + + class MyTool(Tool): """Check and set the agent loop's runtime configuration.""" @@ -79,7 +87,10 @@ class MyTool(Tool): READ_ONLY = frozenset({ "subagents", # observable but replacing it would break the system + "tool_names", + "current_iteration", "_current_iteration", # updated by runner only + "_last_usage", "exec_config", # inspect allowed (e.g. check sandbox), modify blocked "web_config", # inspect allowed (e.g. check enable), modify blocked "model_presets", # config-derived catalog; changes require config reload @@ -103,13 +114,6 @@ class MyTool(Tool): "private_key", "access_token", "refresh_token", "auth", }) - @classmethod - def _is_sensitive_field_name(cls, name: str) -> bool: - lowered = name.lower() - return lowered in cls._SENSITIVE_NAMES or any( - part in cls._SENSITIVE_NAMES for part in lowered.split("_") - ) - RESTRICTED: dict[str, dict[str, Any]] = { "max_iterations": {"type": int, "min": 1, "max": 100}, "context_window_tokens": {"type": int, "min": 4096, "max": 1_000_000}, @@ -123,15 +127,15 @@ class MyTool(Tool): "context_window_tokens", }) - def __init__(self, runtime_state: RuntimeState, modify_allowed: bool = True) -> None: - self._runtime_state = runtime_state + def __init__(self, runtime_control: RuntimeControl, modify_allowed: bool = True) -> None: + self._runtime_control = runtime_control self._modify_allowed = modify_allowed def __deepcopy__(self, memo: dict[int, Any]) -> MyTool: cls = self.__class__ result = cls.__new__(cls) memo[id(self)] = result - result._runtime_state = self._runtime_state + result._runtime_control = self._runtime_control result._modify_allowed = self._modify_allowed return result @@ -208,9 +212,12 @@ class MyTool(Tool): # Path resolution # ------------------------------------------------------------------ - def _resolve_path(self, path: str) -> tuple[Any, str | None]: + def _resolve_path( + self, + snapshot: RuntimeSnapshot, + path: str, + ) -> tuple[object | None, str | None]: parts = path.split(".") - obj: Any = self._runtime_state for part in parts: if part in self._DENIED_ATTRS or part.startswith("__"): return None, f"'{part}' is not accessible" @@ -218,17 +225,13 @@ class MyTool(Tool): return None, f"'{part}' is not accessible" if part.lower() in self._SENSITIVE_NAMES: return None, f"'{part}' is not accessible" - try: - if isinstance(obj, Mapping): - mapping = cast(Mapping[str, Any], obj) - if part in mapping: - obj = mapping[part] - else: - return None, f"'{part}' not found in mapping" - else: - obj = getattr(obj, part) - except (KeyError, AttributeError) as e: - return None, f"'{part}' not found: {e}" + obj: object = snapshot.as_mapping() + for part in parts: + if not _is_string_mapping(obj): + return None, f"'{part}' not found" + if part not in obj: + return None, f"'{part}' not found in mapping" + obj = obj[part] return obj, None @staticmethod @@ -242,20 +245,48 @@ class MyTool(Tool): # ------------------------------------------------------------------ @staticmethod - def _format_status(st: "SubagentStatus", indent: str = " ") -> str: - elapsed = time.monotonic() - st.started_at - tool_summary = ", ".join( - f"{e.get('name', '?')}({e.get('status', '?')})" for e in st.tool_events[-5:] - ) or "none" + def _format_status( + st: "SubagentStatus | Mapping[str, object]", + indent: str = " ", + ) -> str: + if isinstance(st, Mapping): + started_at = st.get("started_at", time.monotonic()) + raw_events = st.get("tool_events", []) + phase = st.get("phase", "unknown") + iteration = st.get("iteration", 0) + usage = st.get("usage", {}) + error = st.get("error") + stop_reason = st.get("stop_reason") + else: + started_at = st.started_at + raw_events = st.tool_events + phase = st.phase + iteration = st.iteration + usage = st.usage + error = st.error + stop_reason = st.stop_reason + elapsed = time.monotonic() - ( + float(started_at) if isinstance(started_at, (int, float)) else time.monotonic() + ) + tool_events = cast(list[object], raw_events) if isinstance(raw_events, list) else [] + tool_summaries: list[str] = [] + for raw_event in tool_events[-5:]: + if not isinstance(raw_event, Mapping): + continue + event = cast(Mapping[str, object], raw_event) + tool_summaries.append( + f"{event.get('name', '?')}({event.get('status', '?')})" + ) + tool_summary = ", ".join(tool_summaries) or "none" lines = [ - f"{indent}phase: {st.phase}, iteration: {st.iteration}, elapsed: {elapsed:.1f}s", + f"{indent}phase: {phase}, iteration: {iteration}, elapsed: {elapsed:.1f}s", f"{indent}tools: {tool_summary}", - f"{indent}usage: {st.usage or 'n/a'}", + f"{indent}usage: {usage or 'n/a'}", ] - if st.error: - lines.append(f"{indent}error: {st.error}") - if st.stop_reason: - lines.append(f"{indent}stop_reason: {st.stop_reason}") + if error: + lines.append(f"{indent}error: {error}") + if stop_reason: + lines.append(f"{indent}stop_reason: {stop_reason}") return "\n".join(lines) @staticmethod @@ -264,29 +295,38 @@ class MyTool(Tool): header = f"Subagent [{val.task_id}] '{val.label}'" detail = MyTool._format_status(val, " ") return f"{header}\n task: {val.task_description}\n{detail}" - # SubagentManager: delegate to its _task_statuses dict - task_statuses = getattr(val, "_task_statuses", None) - if isinstance(task_statuses, dict): - return MyTool._format_value(task_statuses, key) + if _is_subagent_status_snapshot(val): + header = f"Subagent [{val['task_id']}] '{val['label']}'" + detail = MyTool._format_status(val, " ") + return f"{header}\n task: {val['task_description']}\n{detail}" if isinstance(val, Mapping): mapping = cast(Mapping[object, object], val) else: mapping = None + if mapping and set(mapping) == {"_task_statuses"}: + task_statuses = mapping["_task_statuses"] + if isinstance(task_statuses, Mapping): + return MyTool._format_value(task_statuses, key) if ( mapping - and _is_subagent_status(next(iter(mapping.values()))) + and ( + _is_subagent_status(next(iter(mapping.values()))) + or _is_subagent_status_snapshot(next(iter(mapping.values()))) + ) ): - status_mapping: Mapping[object, SubagentStatus] = cast(Any, mapping) prefix = f"{key}: " if key else "" - lines = [f"{prefix}{len(status_mapping)} subagent(s):"] - for tid, st in status_mapping.items(): - detail = MyTool._format_status(st, " ") - lines.append(f" [{tid}] '{st.label}'\n{detail}") + lines = [f"{prefix}{len(mapping)} subagent(s):"] + for tid, st in mapping.items(): + if _is_subagent_status(st): + detail = MyTool._format_status(st, " ") + label = st.label + elif _is_subagent_status_snapshot(st): + detail = MyTool._format_status(st, " ") + label = st.get("label", "?") + else: + continue + lines.append(f" [{tid}] '{label}'\n{detail}") return "\n".join(lines) - dynamic_value = cast(Any, val) - if hasattr(dynamic_value, "tool_names"): - tool_names: Any = getattr(dynamic_value, "tool_names") - return f"tools: {len(tool_names)} registered — {tool_names}" # Scalar types — repr is fine if isinstance(val, (str, int, float, bool, type(None))): r = repr(val) @@ -311,32 +351,6 @@ class MyTool(Tool): return f"{key}: [{len(sequence)} items]" if key else f"[{len(sequence)} items]" r = repr(sequence) return f"{key}: {r}" if key else r - # Complex object — small Pydantic models: show values; others: show field names for navigation - value_type = type(cast(object, val)) - cls_name = value_type.__name__ - model_fields = cast(object, getattr(value_type, "model_fields", None)) - if isinstance(model_fields, Mapping) and model_fields: - fields = list(cast(Mapping[str, object], model_fields).keys()) - if len(fields) <= 8: - # Small config objects: show field=value pairs - pairs: list[str] = [] - for f in fields: - fv = getattr(val, f, "?") - if MyTool._is_sensitive_field_name(f): - continue - if isinstance(fv, (str, int, float, bool, type(None))): - pairs.append(f"{f}={fv!r}") - else: - pairs.append(f"{f}=<{type(fv).__name__}>") - preview = ", ".join(pairs) - return f"{key}: {preview}" if key else preview - else: - attributes = cast(dict[str, Any], getattr(val, "__dict__", {})) - fields = [name for name in attributes if not name.startswith("__")] - if fields: - preview = ", ".join(str(f) for f in fields[:20]) - suffix = ", ..." if len(fields) > 20 else "" - return f"{key}: <{cls_name}> [{preview}{suffix}]" if key else f"<{cls_name}> [{preview}{suffix}]" r = repr(val) return f"{key}: {r}" if key else r @@ -366,7 +380,12 @@ class MyTool(Tool): runtime = request_ctx.runtime if request_ctx is not None else None if runtime is None or key not in self._MODEL_RUNTIME_FIELDS: return False, None - return True, getattr(runtime, key) + values: dict[str, object] = { + "model": runtime.model, + "model_preset": runtime.model_preset, + "context_window_tokens": runtime.context_window_tokens, + } + return True, values[key] def _inspect(self, key: str | None) -> str: if not key: @@ -375,62 +394,64 @@ class MyTool(Tool): request_ctx = current_request_context() if request_ctx is None: return ToolResult.error("Error: current request context is unavailable") + request_values: dict[str, str | None] = { + "channel": request_ctx.channel, + "chat_id": request_ctx.chat_id, + "sender_id": request_ctx.sender_id, + } if key == "request": - return self._format_value( - {field: getattr(request_ctx, field) for field in self._REQUEST_FIELDS}, - key, - ) + return self._format_value(request_values, key) field = key.removeprefix("request.") if field not in self._REQUEST_FIELDS: return ToolResult.error(f"Error: '{key}' not found") - return self._format_value(getattr(request_ctx, field), key) + return self._format_value(request_values[field], key) if "." not in key: found, value = self._current_runtime_value(key) if found: return self._format_value(value, key) + snapshot = self._runtime_control.snapshot() top = key.split(".")[0] if top in self._DENIED_ATTRS or top.startswith("__"): return ToolResult.error(f"Error: '{top}' is not accessible") - obj, err = self._resolve_path(key) + obj, err = self._resolve_path(snapshot, key) if err: - # "scratchpad" alias for _runtime_vars if key == "scratchpad": - rv = self._runtime_state._runtime_vars - return self._format_value(rv, "scratchpad") if rv else "scratchpad is empty" - # Fallback: check _runtime_vars for simple keys stored by modify - if "." not in key and key in self._runtime_state._runtime_vars: - return self._format_value(self._runtime_state._runtime_vars[key], key) + return ( + self._format_value(snapshot.scratchpad, "scratchpad") + if snapshot.scratchpad + else "scratchpad is empty" + ) + if "." not in key and key in snapshot.scratchpad: + return self._format_value(snapshot.scratchpad[key], key) return ToolResult.error(f"Error: {err}") - # Guard against mock auto-generated attributes - if "." not in key and not _has_real_attr(self._runtime_state, key): - if key in self._runtime_state._runtime_vars: - return self._format_value(self._runtime_state._runtime_vars[key], key) - return ToolResult.error(f"Error: '{key}' not found") return self._format_value(obj, key) def _inspect_all(self) -> str: - state = self._runtime_state + snapshot = self._runtime_control.snapshot() + values = snapshot.as_mapping() parts: list[str] = [] - # RESTRICTED keys for k in self.RESTRICTED: found, value = self._current_runtime_value(k) - parts.append(self._format_value(value if found else getattr(state, k, None), k)) + parts.append(self._format_value(value if found else values[k], k)) found, value = self._current_runtime_value("model_preset") parts.append(self._format_value( - value if found else state.model_preset, + value if found else snapshot.model_preset, "model_preset", )) - # Other useful top-level keys shown in description - for k in ("workspace", "provider_retry_mode", "max_tool_result_chars", "_current_iteration", "web_config", "exec_config", "workspace_sandbox", "subagents"): - if _has_real_attr(state, k): - parts.append(self._format_value(getattr(state, k, None), k)) - # Token usage - usage = state._last_usage - if usage: - parts.append(self._format_value(usage, "_last_usage")) - rv = state._runtime_vars - if rv: - parts.append(self._format_value(rv, "scratchpad")) + for k in ( + "workspace", + "provider_retry_mode", + "max_tool_result_chars", + "_current_iteration", + "web_config", + "exec_config", + "subagents", + ): + parts.append(self._format_value(values[k], k)) + if snapshot.last_usage: + parts.append(self._format_value(snapshot.last_usage, "_last_usage")) + if snapshot.scratchpad: + parts.append(self._format_value(snapshot.scratchpad, "scratchpad")) return "\n".join(parts) # -- modify -- @@ -454,48 +475,49 @@ class MyTool(Tool): if leaf.lower() in self._SENSITIVE_NAMES: self._audit("modify", f"BLOCKED sensitive leaf '{leaf}'") return ToolResult.error(f"Error: '{leaf}' is not accessible") - parent, err = self._resolve_path(parent_path) + snapshot = self._runtime_control.snapshot() + _parent, err = self._resolve_path(snapshot, parent_path) if err: return ToolResult.error(f"Error: {err}") - if isinstance(parent, dict): - parent[leaf] = value - else: - setattr(parent, leaf, value) - self._audit("modify", f"{key} = {value!r}") - return f"Set {key} = {value!r}" + self._audit("modify", f"READ_ONLY {key}") + return ToolResult.error(f"Error: '{key}' is read-only and cannot be modified") if key == "model_preset": return self._modify_model_preset(value) if key in self.RESTRICTED: return self._modify_restricted(key, value) - return self._modify_free(key, value) + if key in RUNTIME_COMMAND_KEYS: + return self._modify_runtime_setting(key, value) + if key in RUNTIME_SNAPSHOT_KEYS: + self._audit("modify", f"READ_ONLY {key}") + return ToolResult.error(f"Error: '{key}' is read-only and cannot be modified") + return self._modify_scratchpad(key, value) def _modify_model_preset(self, value: Any) -> str: if not isinstance(value, str) or not value.strip(): return ToolResult.error("Error: 'model_preset' must be a non-empty string") name = value.strip() session_key = current_request_session_key() + old = self._runtime_control.snapshot().model_preset + try: + runtime = self._runtime_control.set_model_preset( + name, + session_key=session_key, + ) + except (KeyError, ValueError) as exc: + message = str(exc.args[0]) if exc.args else str(exc) + punctuation = "" if message.endswith((".", "!", "?")) else "." + return ToolResult.error(f"Error: {message}{punctuation}") if session_key: - try: - runtime = self._runtime_state.set_session_model_preset( - session_key, - name, - ) - except (KeyError, ValueError) as exc: - message = str(exc.args[0]) if exc.args else str(exc) - punctuation = "" if message.endswith((".", "!", "?")) else "." - return ToolResult.error(f"Error: {message}{punctuation}") self._audit("modify", f"model_preset = {name!r}") return ( f"Set model_preset = {name!r} for the next turn; " f"model will be {runtime.model!r}; " f"context_window_tokens will be {runtime.context_window_tokens!r}" ) - result = self._modify_free("model_preset", name) - if isinstance(result, ToolResult) and result.is_error: - return result if result.endswith((".", "!", "?")) else ToolResult.error(f"{result}.") + self._audit("modify", f"model_preset: {old!r} -> {name!r}") return ( - f"{result}; model is now {self._runtime_state.model!r}; " - f"context_window_tokens is now {self._runtime_state.context_window_tokens!r}" + f"Set model_preset = {name!r} (was {old!r}); model is now {runtime.model!r}; " + f"context_window_tokens is now {runtime.context_window_tokens!r}" ) def _modify_restricted(self, key: str, value: Any) -> str: @@ -508,7 +530,7 @@ class MyTool(Tool): value = expected(value) except (ValueError, TypeError): return ToolResult.error(f"Error: '{key}' must be {expected.__name__}, got {type(value).__name__}") - old = getattr(self._runtime_state, key) + old = self._runtime_control.snapshot().as_mapping()[key] if "min" in spec and value < spec["min"]: return ToolResult.error(f"Error: '{key}' must be >= {spec['min']}") if "max" in spec and value > spec["max"]: @@ -521,41 +543,46 @@ class MyTool(Tool): "during an active session; use a configured model_preset" ) if key == "model": - self._runtime_state.set_runtime_model(cast(str, value)) + self._runtime_control.set_model(cast(str, value)) elif key == "context_window_tokens": - self._runtime_state.set_runtime_context_window(cast(int, value)) + self._runtime_control.set_context_window_tokens(cast(int, value)) else: - setattr(self._runtime_state, key, value) - if key == "max_iterations" and hasattr( - self._runtime_state, - "_sync_subagent_runtime_limits", - ): - self._runtime_state._sync_subagent_runtime_limits() + self._runtime_control.set_max_iterations(cast(int, value)) self._audit("modify", f"{key}: {old!r} -> {value!r}") return f"Set {key} = {value!r} (was {old!r})" - def _modify_free(self, key: str, value: Any) -> str: - if _has_real_attr(self._runtime_state, key): - old = getattr(self._runtime_state, key) - if isinstance(old, (str, int, float, bool)): - old_t: type[Any] = type(old) - new_t = cast(type[Any], type(value)) - if old_t is float and new_t is int: - pass # int → float coercion allowed - elif old_t is not new_t: - self._audit( - "modify", - f"REJECTED type mismatch {key}: expects {old_t.__name__}, got {new_t.__name__}", - ) - return ToolResult.error(f"Error: '{key}' expects {old_t.__name__}, got {new_t.__name__}") - try: - setattr(self._runtime_state, key, value) - except (ValueError, KeyError) as e: - message = str(e.args[0] if isinstance(e, KeyError) and e.args else e).strip('"') - self._audit("modify", f"REJECTED {key}: {message}") - return ToolResult.error(f"Error: {message}") - self._audit("modify", f"{key}: {old!r} -> {value!r}") - return f"Set {key} = {value!r} (was {old!r})" + def _modify_runtime_setting(self, key: str, value: Any) -> str: + old = self._runtime_control.snapshot().as_mapping()[key] + if key == "workspace": + if not isinstance(value, str): + return ToolResult.error( + f"Error: 'workspace' expects str, got {type(value).__name__}" + ) + self._runtime_control.set_workspace_display(value) + self._audit("modify", f"workspace: {old!r} -> {value!r}") + return f"Set workspace = {value!r} (was {old!r})" + old_t = type(old) + new_t = cast(type[Any], type(value)) + if old_t is float and new_t is int: + pass + elif old_t is not new_t: + self._audit( + "modify", + f"REJECTED type mismatch {key}: expects {old_t.__name__}, got {new_t.__name__}", + ) + return ToolResult.error( + f"Error: '{key}' expects {old_t.__name__}, got {new_t.__name__}" + ) + if key == "provider_retry_mode": + self._runtime_control.set_provider_retry_mode(cast(str, value)) + elif key == "max_tool_result_chars": + self._runtime_control.set_max_tool_result_chars(cast(int, value)) + else: + raise AssertionError(f"Unhandled runtime command: {key}") + self._audit("modify", f"{key}: {old!r} -> {value!r}") + return f"Set {key} = {value!r} (was {old!r})" + + def _modify_scratchpad(self, key: str, value: Any) -> str: if callable(value): self._audit("modify", f"REJECTED callable {key}") return ToolResult.error("Error: cannot store callable values") @@ -563,12 +590,16 @@ class MyTool(Tool): if err: self._audit("modify", f"REJECTED {key}: {err}") return ToolResult.error(f"Error: {err}") - if key not in self._runtime_state._runtime_vars and len(self._runtime_state._runtime_vars) >= self._MAX_RUNTIME_KEYS: + try: + self._runtime_control.set_scratchpad( + key, + cast(JsonValue, value), + max_keys=self._MAX_RUNTIME_KEYS, + ) + except ValueError as exc: self._audit("modify", f"REJECTED {key}: max keys ({self._MAX_RUNTIME_KEYS}) reached") - return ToolResult.error(f"Error: scratchpad is full (max {self._MAX_RUNTIME_KEYS} keys). Remove unused keys first.") - old = self._runtime_state._runtime_vars.get(key) - self._runtime_state._runtime_vars[key] = value - self._audit("modify", f"scratchpad.{key}: {old!r} -> {value!r}") + return ToolResult.error(f"Error: {exc}. Remove unused keys first.") + self._audit("modify", f"scratchpad.{key} = {value!r}") return f"Set scratchpad.{key} = {value!r}" @classmethod diff --git a/tests/agent/test_self_model_preset.py b/tests/agent/test_self_model_preset.py index f0a1e1072..520755635 100644 --- a/tests/agent/test_self_model_preset.py +++ b/tests/agent/test_self_model_preset.py @@ -5,6 +5,7 @@ import pytest from nanobot.agent.loop import AgentLoop from nanobot.agent.tools.context import RequestContext, request_context +from nanobot.agent.tools.runtime_control import AgentRuntimeControl from nanobot.agent.tools.self import MyTool from nanobot.bus.queue import MessageBus from nanobot.config.schema import ModelPresetConfig @@ -34,6 +35,13 @@ def _make_loop(tmp_path, presets=None, active_preset=None): ) +def _my_tool(loop: AgentLoop) -> MyTool: + return MyTool( + runtime_control=AgentRuntimeControl(loop), + modify_allowed=True, + ) + + def test_model_preset_getter_none_when_not_set(tmp_path) -> None: loop = _make_loop(tmp_path) assert loop.model_preset is None @@ -240,7 +248,7 @@ def test_self_tool_inspect_shows_model_preset(tmp_path) -> None: "fast": ModelPresetConfig(model="openai/gpt-4.1"), } loop = _make_loop(tmp_path, presets=presets, active_preset="fast") - tool = MyTool(runtime_state=loop, modify_allowed=True) + tool = _my_tool(loop) output = tool._inspect_all() assert "model_preset: 'fast'" in output @@ -250,7 +258,7 @@ def test_self_tool_set_model_preset_via_modify(tmp_path) -> None: "fast": ModelPresetConfig(model="openai/gpt-4.1"), } loop = _make_loop(tmp_path, presets=presets) - tool = MyTool(runtime_state=loop, modify_allowed=True) + tool = _my_tool(loop) result = tool._modify("model_preset", "fast") assert "Error" not in result assert loop.model_preset == "fast" @@ -263,7 +271,7 @@ def test_self_tool_set_model_preset_switches_back_to_default(tmp_path) -> None: "fast": ModelPresetConfig(model="openai/gpt-4.1", context_window_tokens=32_768), } loop = _make_loop(tmp_path, presets=presets, active_preset="fast") - tool = MyTool(runtime_state=loop, modify_allowed=True) + tool = _my_tool(loop) result = tool._modify("model_preset", "default") @@ -280,7 +288,7 @@ def test_self_tool_set_model_preset_unknown_lists_available(tmp_path) -> None: "fast": ModelPresetConfig(model="openai/gpt-4.1"), } loop = _make_loop(tmp_path, presets=presets) - tool = MyTool(runtime_state=loop, modify_allowed=True) + tool = _my_tool(loop) result = tool._modify("model_preset", "missing") @@ -295,7 +303,7 @@ def test_self_tool_sets_model_preset_for_current_session(tmp_path) -> None: "fast": ModelPresetConfig(model="openai/gpt-4.1"), } loop = _make_loop(tmp_path, presets=presets) - tool = MyTool(runtime_state=loop, modify_allowed=True) + tool = _my_tool(loop) with request_context(RequestContext( channel="cli", @@ -318,7 +326,7 @@ def test_self_tool_reports_session_preset_provider_configuration_error(tmp_path) loop.set_session_model_preset = MagicMock( side_effect=ValueError("No API key configured for provider 'openai'.") ) - tool = MyTool(runtime_state=loop, modify_allowed=True) + tool = _my_tool(loop) with request_context(RequestContext( channel="cli", @@ -343,7 +351,7 @@ def test_self_tool_rejects_instance_runtime_changes_in_session( value: object, ) -> None: loop = _make_loop(tmp_path) - tool = MyTool(runtime_state=loop, modify_allowed=True) + tool = _my_tool(loop) session = loop.sessions.get_or_create("cli:one") with request_context(RequestContext( @@ -366,7 +374,7 @@ def test_self_tool_set_model_clears_active_preset(tmp_path) -> None: "fast": ModelPresetConfig(model="openai/gpt-4.1"), } loop = _make_loop(tmp_path, presets=presets, active_preset="fast") - tool = MyTool(runtime_state=loop, modify_allowed=True) + tool = _my_tool(loop) result = tool._modify("model", "anthropic/claude-opus-4-5") assert "Error" not in result assert loop.model_preset is None diff --git a/tests/agent/tools/test_runtime_control.py b/tests/agent/tools/test_runtime_control.py new file mode 100644 index 000000000..984934287 --- /dev/null +++ b/tests/agent/tools/test_runtime_control.py @@ -0,0 +1,225 @@ +"""Contract and security regressions for the MyTool runtime boundary.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from nanobot.agent.loop import AgentLoop +from nanobot.agent.tools.runtime_control import ( + RUNTIME_COMMAND_KEYS, + RUNTIME_SNAPSHOT_KEYS, + AgentRuntimeControl, + RuntimeControl, +) +from nanobot.agent.tools.self import MyTool, MyToolConfig +from nanobot.bus.queue import MessageBus +from nanobot.config.schema import ToolsConfig + + +def _make_loop(tmp_path: Path, *, allow_set: bool = False) -> AgentLoop: + provider = MagicMock() + provider.get_default_model.return_value = "test-model" + tools_config = ToolsConfig(my=MyToolConfig(allow_set=allow_set)) + return AgentLoop( + bus=MessageBus(), + provider=provider, + workspace=tmp_path, + model="test-model", + tools_config=tools_config, + ) + + +def _my_tool(loop: AgentLoop) -> MyTool: + tool = loop.tools.get("my") + assert isinstance(tool, MyTool) + return tool + + +def test_agent_loop_assembles_my_tool_with_runtime_control(tmp_path: Path) -> None: + loop = _make_loop(tmp_path) + tool = _my_tool(loop) + + assert isinstance(tool._runtime_control, RuntimeControl) + assert isinstance(tool._runtime_control, AgentRuntimeControl) + assert tool._runtime_control is not loop + assert not hasattr(tool, "_runtime_state") + + +def test_runtime_snapshot_has_exact_allowlist_and_redacts_secrets(tmp_path: Path) -> None: + loop = _make_loop(tmp_path) + loop.web_config.search.api_key = "search-secret" + loop.web_config.proxy = "http://proxy-user:proxy-secret@proxy.example" + loop.unlisted_secret = "loop-secret" + + snapshot = _my_tool(loop)._runtime_control.snapshot() + values = snapshot.as_mapping() + + assert frozenset(values) == RUNTIME_SNAPSHOT_KEYS + assert RUNTIME_COMMAND_KEYS == frozenset({ + "model", + "model_preset", + "max_iterations", + "context_window_tokens", + "provider_retry_mode", + "max_tool_result_chars", + "workspace", + }) + assert "provider" not in values + assert "sessions" not in values + assert "restrict_to_workspace" not in values + assert "unlisted_secret" not in values + rendered = repr(values) + assert "search-secret" not in rendered + assert "proxy-secret" not in rendered + assert "loop-secret" not in rendered + assert snapshot.web_config["proxy"] == "" + + +def test_runtime_snapshot_is_detached_from_mutable_config(tmp_path: Path) -> None: + loop = _make_loop(tmp_path) + control = _my_tool(loop)._runtime_control + snapshot = control.snapshot() + search = snapshot.web_config["search"] + assert isinstance(search, dict) + + search["provider"] = "mutated" + snapshot.exec_config["allow_patterns"] = ["mutated"] + snapshot.tool_names.append("mutated") + + refreshed = control.snapshot() + refreshed_search = refreshed.web_config["search"] + assert isinstance(refreshed_search, dict) + assert refreshed_search["provider"] == loop.web_config.search.provider + assert refreshed.exec_config["allow_patterns"] == loop.exec_config.allow_patterns + assert "mutated" not in refreshed.tool_names + + +@pytest.mark.asyncio +async def test_unlisted_loop_attributes_cannot_be_read_or_modified(tmp_path: Path) -> None: + loop = _make_loop(tmp_path, allow_set=True) + loop.unlisted_control_plane = "internal-secret" + original_workspace_root = loop.workspace_scopes.default_workspace + tool = _my_tool(loop) + + inspected = await tool.execute(action="check", key="unlisted_control_plane") + modified = await tool.execute( + action="set", + key="unlisted_control_plane", + value="scratch-value", + ) + nested = await tool.execute( + action="set", + key="workspace_scopes.default_workspace", + value="elsewhere", + ) + + assert "internal-secret" not in inspected + assert "not found" in inspected + assert modified == "Set scratchpad.unlisted_control_plane = 'scratch-value'" + assert loop.unlisted_control_plane == "internal-secret" + assert "Error" in nested + assert loop.workspace_scopes.default_workspace == original_workspace_root + + +@pytest.mark.asyncio +async def test_default_allow_set_and_public_parameter_schema_are_unchanged( + tmp_path: Path, +) -> None: + loop = _make_loop(tmp_path) + tool = _my_tool(loop) + + assert ToolsConfig().my.allow_set is False + assert tool.parameters == { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["check", "set"], + "description": "Action to perform", + }, + "key": { + "type": "string", + "description": ( + "Dot-path for check/set. Examples: 'max_iterations', 'workspace', " + "'provider_retry_mode'. Use 'request.channel', 'request.chat_id', or " + "'request.sender_id' for current routing metadata. Use 'model_preset' " + "to switch named model presets. For check without key, shows all " + "config values." + ), + }, + "value": { + "description": ( + "New value (for set). Type must match target (int for " + "max_iterations/context_window_tokens, str for model/model_preset)." + ), + }, + }, + "required": ["action"], + } + assert "READ-ONLY MODE" in tool.description + result = await tool.execute(action="set", key="max_iterations", value=80) + assert result == "Error: set is disabled (tools.my.allow_set is false)" + assert loop.max_iterations != 80 + + +@pytest.mark.asyncio +async def test_allowlisted_commands_preserve_runtime_side_effects(tmp_path: Path) -> None: + loop = _make_loop(tmp_path, allow_set=True) + tool = _my_tool(loop) + + max_iterations = await tool.execute( + action="set", + key="max_iterations", + value=80, + ) + retry_mode = await tool.execute( + action="set", + key="provider_retry_mode", + value="persistent", + ) + scratchpad = await tool.execute( + action="set", + key="preference", + value={"concise": True}, + ) + + assert max_iterations == "Set max_iterations = 80 (was 200)" + assert retry_mode == "Set provider_retry_mode = 'persistent' (was 'standard')" + assert scratchpad == "Set scratchpad.preference = {'concise': True}" + assert loop.max_iterations == 80 + assert loop.subagents.max_iterations == 80 + assert loop.provider_retry_mode == "persistent" + assert tool._runtime_control.snapshot().scratchpad == { + "preference": {"concise": True}, + } + + +@pytest.mark.asyncio +async def test_registry_exposes_unchanged_my_tool_actions(tmp_path: Path) -> None: + loop = _make_loop(tmp_path, allow_set=True) + + checked = await loop.tools.execute("my", {"action": "check", "key": "model"}) + changed = await loop.tools.execute( + "my", + {"action": "set", "key": "max_iterations", "value": 80}, + ) + + assert checked == "model: 'test-model'" + assert changed == "Set max_iterations = 80 (was 200)" + assert loop.max_iterations == 80 + + +@pytest.mark.asyncio +async def test_workspace_display_command_cannot_change_path_enforcement(tmp_path: Path) -> None: + loop = _make_loop(tmp_path, allow_set=True) + tool = _my_tool(loop) + + result = await tool.execute(action="set", key="workspace", value="elsewhere") + + assert "Set workspace" in result + assert tool._runtime_control.snapshot().workspace == "elsewhere" + assert loop.workspace == tmp_path + assert loop.workspace_scopes.default_workspace == tmp_path diff --git a/tests/agent/tools/test_self_tool.py b/tests/agent/tools/test_self_tool.py index 449c51495..6f5cffa24 100644 --- a/tests/agent/tools/test_self_tool.py +++ b/tests/agent/tools/test_self_tool.py @@ -8,10 +8,12 @@ from types import MappingProxyType from unittest.mock import MagicMock, patch import pytest -from pydantic import BaseModel from nanobot.agent.tools.context import RequestContext, request_context +from nanobot.agent.tools.runtime_control import AgentRuntimeControl from nanobot.agent.tools.self import MyTool +from nanobot.agent.tools.shell import ExecToolConfig +from nanobot.agent.tools.web import WebSearchConfig, WebToolsConfig from nanobot.config.schema import ModelPresetConfig # --------------------------------------------------------------------------- @@ -27,13 +29,16 @@ def _make_mock_loop(**overrides): loop.workspace = Path("/tmp/workspace") loop.restrict_to_workspace = False loop._start_time = 1000.0 - loop.exec_config = MagicMock() + loop.exec_config = ExecToolConfig() loop.channels_config = MagicMock() loop._last_usage = {"prompt_tokens": 100, "completion_tokens": 50} - loop._runtime_vars = {} + loop.last_usage = loop._last_usage loop._current_iteration = 0 + loop.current_iteration = loop._current_iteration loop.provider_retry_mode = "standard" loop.max_tool_result_chars = 16000 + loop.model_preset = None + loop.model_presets = {} loop._concurrency_gate = None loop._unified_session = False loop._extra_hooks = [] @@ -45,9 +50,7 @@ def _make_mock_loop(**overrides): ) # web_config mock — needed for check tests - loop.web_config = MagicMock() - loop.web_config.enable = True - loop.web_config.search = MagicMock() + loop.web_config = WebToolsConfig() loop.web_config.search.api_key = "sk-secret-key-12345" # Tools registry mock @@ -55,10 +58,13 @@ def _make_mock_loop(**overrides): loop.tools.tool_names = ["read_file", "write_file", "exec", "web_search", "self"] loop.tools.has.side_effect = lambda n: n in loop.tools.tool_names loop.tools.get.return_value = None + loop.tool_names = loop.tools.tool_names # SubagentManager mock loop.subagents = MagicMock() loop.subagents._running_tasks = {"abc123": MagicMock(done=MagicMock(return_value=False))} + loop.subagents._task_statuses = {} + loop.subagents.runtime_statuses.side_effect = lambda: loop.subagents._task_statuses loop.subagents.get_running_count = MagicMock(return_value=1) for k, v in overrides.items(): @@ -67,10 +73,10 @@ def _make_mock_loop(**overrides): return loop -def _make_tool(runtime_state=None): - if runtime_state is None: - runtime_state = _make_mock_loop() - return MyTool(runtime_state=runtime_state) +def _make_tool(loop=None): + if loop is None: + loop = _make_mock_loop() + return MyTool(runtime_control=AgentRuntimeControl(loop)) # --------------------------------------------------------------------------- @@ -87,10 +93,10 @@ class TestInspectSummary: assert "context_window_tokens: 65536" in result @pytest.mark.asyncio - async def test_inspect_includes_runtime_vars(self): + async def test_inspect_includes_scratchpad(self): loop = _make_mock_loop() - loop._runtime_vars = {"task": "review"} - tool = _make_tool(runtime_state=loop) + tool = _make_tool(loop=loop) + tool._runtime_control.set_scratchpad("task", "review", max_keys=64) result = await tool.execute(action="check") assert "task" in result @@ -150,9 +156,7 @@ class TestInspectPathNavigation: @pytest.mark.asyncio async def test_inspect_config_subfield(self): loop = _make_mock_loop() - loop.web_config = MagicMock() - loop.web_config.enable = True - tool = _make_tool(runtime_state=loop) + tool = _make_tool(loop=loop) result = await tool.execute(action="check", key="web_config.enable") assert "True" in result @@ -160,7 +164,7 @@ class TestInspectPathNavigation: async def test_inspect_dict_key_via_dotpath(self): loop = _make_mock_loop() loop._last_usage = {"prompt_tokens": 100, "completion_tokens": 50} - tool = _make_tool(runtime_state=loop) + tool = _make_tool(loop=loop) result = await tool.execute(action="check", key="_last_usage.prompt_tokens") assert "100" in result @@ -179,20 +183,16 @@ class TestInspectPathNavigation: @pytest.mark.asyncio async def test_inspect_nested_config_redacts_sensitive_scalar_fields(self): - class SearchConfig(BaseModel): - provider: str = "tavily" - api_key: str = "sk-test-secret" - base_url: str = "" - max_results: int = 5 - loop = _make_mock_loop() - loop.web_config = MagicMock() - loop.web_config.search = SearchConfig() + loop.web_config.search = WebSearchConfig( + provider="tavily", + api_key="sk-test-secret", + ) tool = _make_tool(loop) result = await tool.execute(action="check", key="web_config.search") - assert "provider='tavily'" in result + assert "tavily" in result assert "sk-test-secret" not in result assert "api_key" not in result.lower() @@ -209,14 +209,14 @@ class TestModifyRestricted: tool = _make_tool() result = await tool.execute(action="set", key="max_iterations", value=80) assert "Set max_iterations = 80" in result - assert tool._runtime_state.max_iterations == 80 + assert tool._runtime_control.snapshot().max_iterations == 80 @pytest.mark.asyncio async def test_modify_restricted_out_of_range(self): tool = _make_tool() result = await tool.execute(action="set", key="max_iterations", value=0) assert "Error" in result - assert tool._runtime_state.max_iterations == 40 + assert tool._runtime_control.snapshot().max_iterations == 40 @pytest.mark.asyncio async def test_modify_restricted_max_exceeded(self): @@ -241,12 +241,12 @@ class TestModifyRestricted: tool = _make_tool() result = await tool.execute(action="set", key="max_iterations", value="80") assert "Set max_iterations" in result - assert tool._runtime_state.max_iterations == 80 + assert tool._runtime_control.snapshot().max_iterations == 80 @pytest.mark.asyncio async def test_modify_context_window_valid(self): loop = _make_mock_loop() - tool = _make_tool(runtime_state=loop) + tool = _make_tool(loop=loop) result = await tool.execute(action="set", key="context_window_tokens", value=131072) assert "Set context_window_tokens" in result assert loop.context_window_tokens == 131072 @@ -324,15 +324,15 @@ class TestModifyFree: tool = _make_tool() result = await tool.execute(action="set", key="provider_retry_mode", value="persistent") assert "Set provider_retry_mode" in result - assert tool._runtime_state.provider_retry_mode == "persistent" + assert tool._runtime_control.snapshot().provider_retry_mode == "persistent" @pytest.mark.asyncio - async def test_modify_new_key_stores_in_runtime_vars(self): - """Modifying a non-existing attribute should store in _runtime_vars.""" + async def test_modify_new_key_stores_in_scratchpad(self): + """Modifying an unknown key should store it in the scratchpad.""" tool = _make_tool() result = await tool.execute(action="set", key="my_custom_var", value="hello") assert "my_custom_var" in result - assert tool._runtime_state._runtime_vars["my_custom_var"] == "hello" + assert tool._runtime_control.snapshot().scratchpad["my_custom_var"] == "hello" @pytest.mark.asyncio async def test_modify_rejects_callable(self): @@ -351,14 +351,14 @@ class TestModifyFree: tool = _make_tool() result = await tool.execute(action="set", key="items", value=[1, 2, 3]) assert result == "Set scratchpad.items = [1, 2, 3]" - assert tool._runtime_state._runtime_vars["items"] == [1, 2, 3] + assert tool._runtime_control.snapshot().scratchpad["items"] == [1, 2, 3] @pytest.mark.asyncio async def test_modify_allows_dict(self): tool = _make_tool() result = await tool.execute(action="set", key="data", value={"a": 1}) assert result == "Set scratchpad.data = {'a': 1}" - assert tool._runtime_state._runtime_vars["data"] == {"a": 1} + assert tool._runtime_control.snapshot().scratchpad["data"] == {"a": 1} @pytest.mark.asyncio async def test_modify_whitespace_key_rejected(self): @@ -396,7 +396,7 @@ class TestModifyFree: result = await tool.execute(action="set", key="provider_retry_mode", value=42) assert "Error" in result assert "str" in result - assert tool._runtime_state.provider_retry_mode == "standard" + assert tool._runtime_control.snapshot().provider_retry_mode == "standard" @pytest.mark.asyncio async def test_modify_existing_int_attr_wrong_type_rejected(self): @@ -404,7 +404,7 @@ class TestModifyFree: tool = _make_tool() result = await tool.execute(action="set", key="max_tool_result_chars", value="big") assert "Error" in result - assert tool._runtime_state.max_tool_result_chars == 16000 + assert tool._runtime_control.snapshot().max_tool_result_chars == 16000 # --------------------------------------------------------------------------- @@ -486,11 +486,12 @@ class TestModifyOpen: assert "protected" in result @pytest.mark.asyncio - async def test_modify_workspace_allowed(self): - """workspace was READONLY in v1, now freely modifiable.""" + async def test_modify_workspace_preserves_display_compatibility(self): + """The compatibility value is isolated from filesystem security boundaries.""" tool = _make_tool() result = await tool.execute(action="set", key="workspace", value="/new/path") assert "Set workspace" in result + assert tool._runtime_control.snapshot().workspace == "/new/path" @pytest.mark.asyncio async def test_modify_mcp_servers_blocked(self): @@ -584,28 +585,28 @@ class TestUnknownAction: # --------------------------------------------------------------------------- -# runtime_vars limits (from code review) +# scratchpad limits # --------------------------------------------------------------------------- -class TestRuntimeVarsLimits: +class TestScratchpadLimits: @pytest.mark.asyncio - async def test_runtime_vars_rejects_at_max_keys(self): - loop = _make_mock_loop() - loop._runtime_vars = {f"key_{i}": i for i in range(64)} - tool = _make_tool(runtime_state=loop) + async def test_scratchpad_rejects_at_max_keys(self): + tool = _make_tool() + for i in range(64): + tool._runtime_control.set_scratchpad(f"key_{i}", i, max_keys=64) result = await tool.execute(action="set", key="overflow", value="data") assert "full" in result - assert "overflow" not in loop._runtime_vars + assert "overflow" not in tool._runtime_control.snapshot().scratchpad @pytest.mark.asyncio - async def test_runtime_vars_allows_update_existing_key_at_max(self): - loop = _make_mock_loop() - loop._runtime_vars = {f"key_{i}": i for i in range(64)} - tool = _make_tool(runtime_state=loop) + async def test_scratchpad_allows_update_existing_key_at_max(self): + tool = _make_tool() + for i in range(64): + tool._runtime_control.set_scratchpad(f"key_{i}", i, max_keys=64) result = await tool.execute(action="set", key="key_0", value="updated") assert "Error" not in result - assert loop._runtime_vars["key_0"] == "updated" + assert tool._runtime_control.snapshot().scratchpad["key_0"] == "updated" # --------------------------------------------------------------------------- @@ -844,7 +845,7 @@ class TestInspectTaskStatuses: usage={"prompt_tokens": 500, "completion_tokens": 100}, ), } - tool = _make_tool(runtime_state=loop) + tool = _make_tool(loop=loop) result = await tool.execute(action="check", key="subagents._task_statuses") assert "abc12345" in result assert "read logs" in result @@ -865,7 +866,7 @@ class TestInspectTaskStatuses: stop_reason="completed", ) loop.subagents._task_statuses = {"xyz": status} - tool = _make_tool(runtime_state=loop) + tool = _make_tool(loop=loop) result = await tool.execute(action="check", key="subagents._task_statuses.xyz") assert "search code" in result assert "completed" in result @@ -879,7 +880,10 @@ class TestReadOnlyMode: def _make_readonly_tool(self): loop = _make_mock_loop() - return MyTool(runtime_state=loop, modify_allowed=False) + return MyTool( + runtime_control=AgentRuntimeControl(loop), + modify_allowed=False, + ) @pytest.mark.asyncio async def test_inspect_allowed_in_readonly(self): @@ -904,13 +908,13 @@ class TestReadOnlyMode: # --------------------------------------------------------------------------- -# runtime vars check fallback (Fix #1: cross-turn memory) +# scratchpad inspection # --------------------------------------------------------------------------- -class TestRuntimeVarsInspectFallback: +class TestScratchpadInspection: @pytest.mark.asyncio - async def test_inspect_runtime_var_after_modify(self): + async def test_inspect_scratchpad_value_after_modify(self): """Design doc scenario: set then check should return the value.""" tool = _make_tool() await tool.execute(action="set", key="user_prefers_concise", value=True) @@ -918,14 +922,14 @@ class TestRuntimeVarsInspectFallback: assert "True" in result @pytest.mark.asyncio - async def test_inspect_runtime_var_string(self): + async def test_inspect_scratchpad_string(self): tool = _make_tool() await tool.execute(action="set", key="current_project", value="nanobot") result = await tool.execute(action="check", key="current_project") assert "nanobot" in result @pytest.mark.asyncio - async def test_inspect_runtime_var_dict(self): + async def test_inspect_scratchpad_dict(self): tool = _make_tool() await tool.execute(action="set", key="task_meta", value={"step": 2, "total": 5}) result = await tool.execute(action="check", key="task_meta") @@ -958,7 +962,7 @@ class TestSensitiveSubFieldBlocking: loop = _make_mock_loop() loop.some_config = MagicMock() loop.some_config.password = "hunter2" - tool = _make_tool(runtime_state=loop) + tool = _make_tool(loop=loop) result = await tool.execute(action="check", key="some_config.password") assert "not accessible" in result @@ -967,7 +971,7 @@ class TestSensitiveSubFieldBlocking: loop = _make_mock_loop() loop.vault = MagicMock() loop.vault.secret = "classified" - tool = _make_tool(runtime_state=loop) + tool = _make_tool(loop=loop) result = await tool.execute(action="check", key="vault.secret") assert "not accessible" in result @@ -976,7 +980,7 @@ class TestSensitiveSubFieldBlocking: loop = _make_mock_loop() loop.auth_data = MagicMock() loop.auth_data.token = "jwt-payload" - tool = _make_tool(runtime_state=loop) + tool = _make_tool(loop=loop) result = await tool.execute(action="check", key="auth_data.token") assert "not accessible" in result @@ -992,7 +996,7 @@ class TestSensitiveSubFieldBlocking: async def test_modify_password_blocked(self): loop = _make_mock_loop() loop.some_config = MagicMock() - tool = _make_tool(runtime_state=loop) + tool = _make_tool(loop=loop) result = await tool.execute(action="set", key="some_config.password", value="evil") assert "not accessible" in result @@ -1083,8 +1087,8 @@ class TestSecurityAttributeProtection: @pytest.mark.asyncio async def test_modify_model_presets_dotpath_blocked(self): """The config-derived model preset catalog is inspectable but not mutable.""" - presets = {"fast": {"model": "fast-model"}} - tool = _make_tool(runtime_state=_make_mock_loop(model_presets=presets)) + presets = {"fast": ModelPresetConfig(model="fast-model")} + tool = _make_tool(loop=_make_mock_loop(model_presets=presets)) result = await tool.execute( action="set", @@ -1093,14 +1097,14 @@ class TestSecurityAttributeProtection: ) assert "read-only" in result - assert presets == {"fast": {"model": "fast-model"}} + assert presets == {"fast": ModelPresetConfig(model="fast-model")} @pytest.mark.asyncio async def test_inspect_read_only_model_preset_dotpath(self): presets = MappingProxyType({ "fast": ModelPresetConfig(model="fast-model"), }) - tool = _make_tool(runtime_state=_make_mock_loop(model_presets=presets)) + tool = _make_tool(loop=_make_mock_loop(model_presets=presets)) result = await tool.execute(action="check", key="model_presets.fast.model") @@ -1150,7 +1154,8 @@ class TestLastUsageInSummary: async def test_last_usage_not_shown_when_empty(self): loop = _make_mock_loop() loop._last_usage = {} - tool = _make_tool(runtime_state=loop) + loop.last_usage = loop._last_usage + tool = _make_tool(loop=loop) result = await tool.execute(action="check") assert "_last_usage" not in result diff --git a/tests/agent/tools/test_self_tool_runtime_sync.py b/tests/agent/tools/test_self_tool_runtime_sync.py index 8b49dc7c0..45a40d528 100644 --- a/tests/agent/tools/test_self_tool_runtime_sync.py +++ b/tests/agent/tools/test_self_tool_runtime_sync.py @@ -4,23 +4,23 @@ from unittest.mock import MagicMock import pytest +from nanobot.agent.loop import AgentLoop +from nanobot.agent.tools.runtime_control import AgentRuntimeControl from nanobot.agent.tools.self import MyTool +from nanobot.bus.queue import MessageBus @pytest.mark.asyncio -async def test_my_tool_max_iterations_syncs_subagent_limit() -> None: - loop = MagicMock() - loop.max_iterations = 40 - loop._runtime_vars = {} - loop.subagents = MagicMock() - loop.subagents.max_iterations = loop.max_iterations - - def _sync_subagent_runtime_limits() -> None: - loop.subagents.max_iterations = loop.max_iterations - - loop._sync_subagent_runtime_limits = _sync_subagent_runtime_limits - - tool = MyTool(runtime_state=loop) +async def test_my_tool_max_iterations_syncs_subagent_limit(tmp_path) -> None: + provider = MagicMock() + provider.get_default_model.return_value = "test-model" + loop = AgentLoop( + bus=MessageBus(), + provider=provider, + workspace=tmp_path, + max_iterations=40, + ) + tool = MyTool(runtime_control=AgentRuntimeControl(loop)) result = await tool.execute(action="set", key="max_iterations", value=80)