diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index de6496017..6908dc4e5 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -24,12 +24,15 @@ from nanobot.agent.hook import AgentHook, CompositeHook from nanobot.agent.memory import Consolidator from nanobot.agent.progress_hook import AgentProgressHook from nanobot.agent.runner import _MAX_INJECTIONS_PER_TURN, AgentRunner, AgentRunSpec +from nanobot.agent.runtime_model import RuntimeModelCoordinator +from nanobot.agent.streaming import StreamingCoordinator from nanobot.agent.subagent import SubagentManager from nanobot.agent.tools.context import RequestContext, bind_request_context, reset_request_context 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.self import MyTool +from nanobot.agent.turn_session import TurnSessionCoordinator from nanobot.bus.events import InboundMessage, OutboundMessage from nanobot.bus.progress import build_bus_progress_callback from nanobot.bus.queue import MessageBus @@ -160,11 +163,10 @@ class AgentLoop: def llm_runtime(self) -> LLMRuntime: """Return the current provider/model pair owned by this loop.""" - self._refresh_provider_snapshot() - return LLMRuntime(self.provider, self.model) + return self.runtime_models.llm_runtime() - _RUNTIME_CHECKPOINT_KEY = "runtime_checkpoint" - _PENDING_USER_TURN_KEY = "pending_user_turn" + _RUNTIME_CHECKPOINT_KEY = TurnSessionCoordinator.RUNTIME_CHECKPOINT_KEY + _PENDING_USER_TURN_KEY = TurnSessionCoordinator.PENDING_USER_TURN_KEY # Event-driven state transition table. # Handlers return an event string; the driver looks up the next state here. @@ -220,6 +222,7 @@ class AgentLoop: _tc = tools_config or ToolsConfig() defaults = AgentDefaults() self.bus = bus + self.streaming = StreamingCoordinator(bus) self.runtime_events = runtime_events or RuntimeEventBus() self.runtime_event_publisher = RuntimeEventPublisher(self.runtime_events) self.channels_config = channels_config @@ -271,6 +274,7 @@ class AgentLoop: self.context = ContextBuilder(workspace, timezone=timezone, disabled_skills=disabled_skills) self.sessions = session_manager or SessionManager(workspace) + self.turn_sessions = TurnSessionCoordinator(self.sessions) self.tools = ToolRegistry() # One file-read/write tracker per logical session. The tool registry is # shared by this loop, so tools resolve the active state via contextvars. @@ -332,6 +336,7 @@ class AgentLoop: ) self.model_presets: dict[str, ModelPresetConfig] = model_presets or {} self._active_preset: str | None = None + self.runtime_models = RuntimeModelCoordinator(self) if model_preset: self.set_model_preset(model_preset, publish_update=False) self._register_default_tools() @@ -398,7 +403,7 @@ class AgentLoop: def _sync_subagent_runtime_limits(self) -> None: """Keep subagent runtime limits aligned with mutable loop settings.""" - self.subagents.max_iterations = self.max_iterations + self.runtime_models.sync_subagent_runtime_limits() def _apply_provider_snapshot( self, @@ -408,52 +413,14 @@ class AgentLoop: model_preset: str | None = None, ) -> None: """Swap model/provider for future turns without disturbing an active one.""" - provider = snapshot.provider - model = snapshot.model - context_window_tokens = snapshot.context_window_tokens - old_model = self.model - self.provider = provider - self.model = model - self.context_window_tokens = context_window_tokens - self.runner.provider = provider - self.subagents.set_provider(provider, model) - self.consolidator.set_provider(provider, model, context_window_tokens) - self._provider_signature = snapshot.signature - if publish_update and self._runtime_model_publisher is not None: - self._runtime_model_publisher( - self.model, - model_preset if model_preset is not None else self.model_preset, - ) - if publish_update: - self._runtime_events().runtime_model_changed( - self.model, - model_preset if model_preset is not None else self.model_preset, - ) - logger.info("Runtime model switched for next turn: {} -> {}", old_model, model) + self.runtime_models.apply_provider_snapshot( + snapshot, + publish_update=publish_update, + model_preset=model_preset, + ) def _refresh_provider_snapshot(self) -> None: - if self._provider_snapshot_loader is None: - return - try: - snapshot = self._provider_snapshot_loader() - except Exception: - logger.exception("Failed to refresh provider config") - return - default_selection = preset_helpers.default_selection_signature(snapshot.signature) - if self._active_preset and self._default_selection_signature in (None, default_selection): - self._default_selection_signature = default_selection - try: - snapshot = self._build_model_preset_snapshot(self._active_preset) - except Exception: - logger.exception("Failed to refresh active model preset") - return - else: - self._active_preset = None - self._default_selection_signature = default_selection - if snapshot.signature == self._provider_signature: - return - self._default_selection_signature = preset_helpers.default_selection_signature(snapshot.signature) - self._apply_provider_snapshot(snapshot) + self.runtime_models.refresh_provider_snapshot() @property def model_preset(self) -> str | None: @@ -464,19 +431,11 @@ class AgentLoop: self.set_model_preset(name) def _build_model_preset_snapshot(self, name: str) -> ProviderSnapshot: - return preset_helpers.build_runtime_preset_snapshot( - name=name, - presets=self.model_presets, - provider=self.provider, - loader=self._preset_snapshot_loader, - ) + return self.runtime_models.build_model_preset_snapshot(name) def set_model_preset(self, name: str | None, *, publish_update: bool = True) -> None: """Resolve a preset by name and apply all runtime model dependents.""" - name = preset_helpers.normalize_preset_name(name, self.model_presets) - snapshot = self._build_model_preset_snapshot(name) - self._apply_provider_snapshot(snapshot, publish_update=publish_update, model_preset=name) - self._active_preset = name + self.runtime_models.set_model_preset(name, publish_update=publish_update) def _register_default_tools(self) -> None: """Register the default set of tools via plugin loader.""" @@ -969,35 +928,9 @@ class AgentLoop: try: on_stream = on_stream_end = None if msg.metadata.get("_wants_stream"): - # Split one answer into distinct stream segments. - stream_base_id = f"{msg.session_key}:{time.time_ns()}" - stream_segment = 0 - - def _current_stream_id() -> str: - return f"{stream_base_id}:{stream_segment}" - - async def on_stream(delta: str) -> None: - meta = dict(msg.metadata or {}) - meta["_stream_delta"] = True - meta["_stream_id"] = _current_stream_id() - await self.bus.publish_outbound(OutboundMessage( - channel=msg.channel, chat_id=msg.chat_id, - content=delta, - metadata=meta, - )) - - async def on_stream_end(*, resuming: bool = False) -> None: - nonlocal stream_segment - meta = dict(msg.metadata or {}) - meta["_stream_end"] = True - meta["_resuming"] = resuming - meta["_stream_id"] = _current_stream_id() - await self.bus.publish_outbound(OutboundMessage( - channel=msg.channel, chat_id=msg.chat_id, - content="", - metadata=meta, - )) - stream_segment += 1 + callbacks = self.streaming.build_callbacks(msg) + on_stream = callbacks.on_stream + on_stream_end = callbacks.on_stream_end response = await self._process_message( msg, on_stream=on_stream, on_stream_end=on_stream_end, @@ -1697,104 +1630,35 @@ class AgentLoop: def _set_runtime_checkpoint(self, session: Session, payload: dict[str, Any]) -> None: """Persist the latest in-flight turn state into session metadata.""" - session.metadata[self._RUNTIME_CHECKPOINT_KEY] = payload - self.sessions.save(session) + self._turn_session_coordinator().set_runtime_checkpoint(session, payload) + + def _turn_session_coordinator(self) -> TurnSessionCoordinator: + coordinator = getattr(self, "turn_sessions", None) + if coordinator is None: + coordinator = TurnSessionCoordinator(getattr(self, "sessions", None)) + self.turn_sessions = coordinator + return coordinator def _mark_pending_user_turn(self, session: Session) -> None: - session.metadata[self._PENDING_USER_TURN_KEY] = True + self._turn_session_coordinator().mark_pending_user_turn(session) def _clear_pending_user_turn(self, session: Session) -> None: - session.metadata.pop(self._PENDING_USER_TURN_KEY, None) + self._turn_session_coordinator().clear_pending_user_turn(session) def _clear_runtime_checkpoint(self, session: Session) -> None: - if self._RUNTIME_CHECKPOINT_KEY in session.metadata: - session.metadata.pop(self._RUNTIME_CHECKPOINT_KEY, None) + self._turn_session_coordinator().clear_runtime_checkpoint(session) @staticmethod def _checkpoint_message_key(message: dict[str, Any]) -> tuple[Any, ...]: - return ( - message.get("role"), - message.get("content"), - message.get("tool_call_id"), - message.get("name"), - message.get("tool_calls"), - message.get("reasoning_content"), - message.get("thinking_blocks"), - ) + return TurnSessionCoordinator.checkpoint_message_key(message) def _restore_runtime_checkpoint(self, session: Session) -> bool: """Materialize an unfinished turn into session history before a new request.""" - from datetime import datetime - - checkpoint = session.metadata.get(self._RUNTIME_CHECKPOINT_KEY) - if not isinstance(checkpoint, dict): - return False - - assistant_message = checkpoint.get("assistant_message") - completed_tool_results = checkpoint.get("completed_tool_results") or [] - pending_tool_calls = checkpoint.get("pending_tool_calls") or [] - - restored_messages: list[dict[str, Any]] = [] - if isinstance(assistant_message, dict): - restored = dict(assistant_message) - restored.setdefault("timestamp", datetime.now().isoformat()) - restored_messages.append(restored) - for message in completed_tool_results: - if isinstance(message, dict): - restored = dict(message) - restored.setdefault("timestamp", datetime.now().isoformat()) - restored_messages.append(restored) - for tool_call in pending_tool_calls: - if not isinstance(tool_call, dict): - continue - tool_id = tool_call.get("id") - name = ((tool_call.get("function") or {}).get("name")) or "tool" - restored_messages.append( - { - "role": "tool", - "tool_call_id": tool_id, - "name": name, - "content": "Error: Task interrupted before this tool finished.", - "timestamp": datetime.now().isoformat(), - } - ) - - overlap = 0 - max_overlap = min(len(session.messages), len(restored_messages)) - for size in range(max_overlap, 0, -1): - existing = session.messages[-size:] - restored = restored_messages[:size] - if all( - self._checkpoint_message_key(left) == self._checkpoint_message_key(right) - for left, right in zip(existing, restored) - ): - overlap = size - break - session.messages.extend(restored_messages[overlap:]) - - self._clear_pending_user_turn(session) - self._clear_runtime_checkpoint(session) - return True + return self._turn_session_coordinator().restore_runtime_checkpoint(session) def _restore_pending_user_turn(self, session: Session) -> bool: """Close a turn that only persisted the user message before crashing.""" - from datetime import datetime - - if not session.metadata.get(self._PENDING_USER_TURN_KEY): - return False - - if session.messages and session.messages[-1].get("role") == "user": - session.messages.append( - { - "role": "assistant", - "content": "Error: Task interrupted before a response was generated.", - "timestamp": datetime.now().isoformat(), - } - ) - session.updated_at = datetime.now() - - self._clear_pending_user_turn(session) - return True + return self._turn_session_coordinator().restore_pending_user_turn(session) async def process_direct( self, diff --git a/nanobot/agent/runtime_model.py b/nanobot/agent/runtime_model.py new file mode 100644 index 000000000..865a3a185 --- /dev/null +++ b/nanobot/agent/runtime_model.py @@ -0,0 +1,107 @@ +"""Runtime model/provider coordination for the agent loop.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from loguru import logger + +from nanobot.agent import model_presets as preset_helpers +from nanobot.providers.factory import ProviderSnapshot +from nanobot.utils.llm_runtime import LLMRuntime + +if TYPE_CHECKING: + from nanobot.agent.loop import AgentLoop + + +class RuntimeModelCoordinator: + """Owns mutable provider/model state transitions for an AgentLoop.""" + + def __init__(self, loop: AgentLoop) -> None: + self._loop = loop + + def llm_runtime(self) -> LLMRuntime: + """Return the current provider/model pair owned by the loop.""" + self.refresh_provider_snapshot() + return LLMRuntime(self._loop.provider, self._loop.model) + + def sync_subagent_runtime_limits(self) -> None: + """Keep subagent runtime limits aligned with mutable loop settings.""" + self._loop.subagents.max_iterations = self._loop.max_iterations + + def apply_provider_snapshot( + self, + snapshot: ProviderSnapshot, + *, + publish_update: bool = True, + model_preset: str | None = None, + ) -> None: + """Swap model/provider for future turns without disturbing an active one.""" + loop = self._loop + provider = snapshot.provider + model = snapshot.model + context_window_tokens = snapshot.context_window_tokens + old_model = loop.model + loop.provider = provider + loop.model = model + loop.context_window_tokens = context_window_tokens + loop.runner.provider = provider + loop.subagents.set_provider(provider, model) + loop.consolidator.set_provider(provider, model, context_window_tokens) + loop._provider_signature = snapshot.signature + active_preset = model_preset if model_preset is not None else loop.model_preset + if publish_update and loop._runtime_model_publisher is not None: + loop._runtime_model_publisher(loop.model, active_preset) + if publish_update: + loop._runtime_events().runtime_model_changed(loop.model, active_preset) + logger.info("Runtime model switched for next turn: {} -> {}", old_model, model) + + def refresh_provider_snapshot(self) -> None: + """Refresh runtime provider state from the configured snapshot loader.""" + loop = self._loop + if loop._provider_snapshot_loader is None: + return + try: + snapshot = loop._provider_snapshot_loader() + except Exception: + logger.exception("Failed to refresh provider config") + return + default_selection = preset_helpers.default_selection_signature(snapshot.signature) + if loop._active_preset and loop._default_selection_signature in (None, default_selection): + loop._default_selection_signature = default_selection + try: + snapshot = self.build_model_preset_snapshot(loop._active_preset) + except Exception: + logger.exception("Failed to refresh active model preset") + return + else: + loop._active_preset = None + loop._default_selection_signature = default_selection + if snapshot.signature == loop._provider_signature: + return + loop._default_selection_signature = preset_helpers.default_selection_signature( + snapshot.signature + ) + self.apply_provider_snapshot(snapshot) + + def build_model_preset_snapshot(self, name: str) -> ProviderSnapshot: + """Resolve a preset into a provider snapshot.""" + loop = self._loop + return preset_helpers.build_runtime_preset_snapshot( + name=name, + presets=loop.model_presets, + provider=loop.provider, + loader=loop._preset_snapshot_loader, + ) + + def set_model_preset(self, name: str | None, *, publish_update: bool = True) -> None: + """Resolve a preset by name and apply all runtime model dependents.""" + loop = self._loop + normalized = preset_helpers.normalize_preset_name(name, loop.model_presets) + snapshot = self.build_model_preset_snapshot(normalized) + self.apply_provider_snapshot( + snapshot, + publish_update=publish_update, + model_preset=normalized, + ) + loop._active_preset = normalized diff --git a/nanobot/agent/streaming.py b/nanobot/agent/streaming.py new file mode 100644 index 000000000..292ffd2a7 --- /dev/null +++ b/nanobot/agent/streaming.py @@ -0,0 +1,62 @@ +"""Streaming response callbacks for bus-backed channels.""" + +from __future__ import annotations + +import time +from collections.abc import Awaitable, Callable +from dataclasses import dataclass + +from nanobot.bus.events import InboundMessage, OutboundMessage +from nanobot.bus.queue import MessageBus + + +@dataclass(frozen=True) +class StreamCallbacks: + on_stream: Callable[[str], Awaitable[None]] + on_stream_end: Callable[..., Awaitable[None]] + + +class StreamingCoordinator: + """Builds outbound bus callbacks for segmented response streaming.""" + + def __init__(self, bus: MessageBus) -> None: + self._bus = bus + + def build_callbacks(self, msg: InboundMessage) -> StreamCallbacks: + """Split one answer into stream segments and publish deltas to the bus.""" + stream_base_id = f"{msg.session_key}:{time.time_ns()}" + stream_segment = 0 + + def _current_stream_id() -> str: + return f"{stream_base_id}:{stream_segment}" + + async def on_stream(delta: str) -> None: + meta = dict(msg.metadata or {}) + meta["_stream_delta"] = True + meta["_stream_id"] = _current_stream_id() + await self._bus.publish_outbound( + OutboundMessage( + channel=msg.channel, + chat_id=msg.chat_id, + content=delta, + metadata=meta, + ) + ) + + async def on_stream_end(*, resuming: bool = False) -> None: + nonlocal stream_segment + meta = dict(msg.metadata or {}) + meta["_stream_end"] = True + meta["_resuming"] = resuming + meta["_stream_id"] = _current_stream_id() + await self._bus.publish_outbound( + OutboundMessage( + channel=msg.channel, + chat_id=msg.chat_id, + content="", + metadata=meta, + ) + ) + stream_segment += 1 + + return StreamCallbacks(on_stream=on_stream, on_stream_end=on_stream_end) diff --git a/nanobot/agent/tools/cli_apps.py b/nanobot/agent/tools/cli_apps.py index e794c09d6..0d6376fea 100644 --- a/nanobot/agent/tools/cli_apps.py +++ b/nanobot/agent/tools/cli_apps.py @@ -5,22 +5,17 @@ from __future__ import annotations from pathlib import Path from typing import Any -from pydantic import Field - from nanobot.agent.tools.base import Tool, tool_parameters -from nanobot.agent.tools.schema import ArraySchema, BooleanSchema, IntegerSchema, StringSchema, tool_parameters_schema -from nanobot.security.workspace_access import current_tool_workspace +from nanobot.agent.tools.schema import ( + ArraySchema, + BooleanSchema, + IntegerSchema, + StringSchema, + tool_parameters_schema, +) from nanobot.apps.cli import CliAppError, CliAppManager, CliAppsRuntimeConfig -from nanobot.config_base import Base - - -class CliAppsToolConfig(Base): - """CLI Apps tool configuration.""" - - enable: bool = True - install_timeout: int = Field(default=300, ge=1, le=3600) - run_timeout: int = Field(default=60, ge=1, le=600) - catalog_ttl_seconds: int = Field(default=3600, ge=60, le=86_400) +from nanobot.config.tool_configs import CliAppsToolConfig +from nanobot.security.workspace_access import current_tool_workspace @tool_parameters( diff --git a/nanobot/agent/tools/filesystem.py b/nanobot/agent/tools/filesystem.py index 9c1854217..26e04698b 100644 --- a/nanobot/agent/tools/filesystem.py +++ b/nanobot/agent/tools/filesystem.py @@ -16,17 +16,11 @@ from nanobot.agent.tools.schema import ( StringSchema, tool_parameters_schema, ) -from nanobot.config_base import Base +from nanobot.config.tool_configs import FileToolsConfig from nanobot.security.workspace_access import current_tool_workspace from nanobot.utils.helpers import build_image_content_blocks, detect_image_mime -class FileToolsConfig(Base): - """Filesystem tools configuration.""" - - enable: bool = True # built-in file tools on by default - - class _FsTool(Tool): """Shared base for filesystem tools — common init and path resolution.""" diff --git a/nanobot/agent/tools/image_generation.py b/nanobot/agent/tools/image_generation.py index 82a09fb91..2cdaf29e3 100644 --- a/nanobot/agent/tools/image_generation.py +++ b/nanobot/agent/tools/image_generation.py @@ -5,8 +5,6 @@ from __future__ import annotations from pathlib import Path from typing import TYPE_CHECKING, Any -from pydantic import Field - from nanobot.agent.tools.base import Tool, tool_parameters from nanobot.agent.tools.schema import ( ArraySchema, @@ -14,14 +12,14 @@ from nanobot.agent.tools.schema import ( StringSchema, tool_parameters_schema, ) -from nanobot.security.workspace_access import current_tool_workspace from nanobot.config.paths import get_media_dir -from nanobot.config_base import Base +from nanobot.config.tool_configs import ImageGenerationToolConfig from nanobot.providers.image_generation import ( ImageGenerationError, ImageGenerationProvider, get_image_gen_provider, ) +from nanobot.security.workspace_access import current_tool_workspace from nanobot.security.workspace_policy import WorkspaceBoundaryError, resolve_allowed_path from nanobot.utils.artifacts import ( ArtifactError, @@ -34,17 +32,6 @@ if TYPE_CHECKING: from nanobot.config.schema import ProviderConfig -class ImageGenerationToolConfig(Base): - """Image generation tool configuration.""" - enabled: bool = False - provider: str = "openrouter" - model: str = "openai/gpt-5.4-image-2" - default_aspect_ratio: str = "1:1" - default_image_size: str = "1K" - max_images_per_turn: int = Field(default=4, ge=1, le=8) - save_dir: str = "generated" - - @tool_parameters( tool_parameters_schema( prompt=StringSchema( diff --git a/nanobot/agent/tools/self.py b/nanobot/agent/tools/self.py index 1e60b3457..fa607705d 100644 --- a/nanobot/agent/tools/self.py +++ b/nanobot/agent/tools/self.py @@ -10,18 +10,12 @@ from loguru import logger from nanobot.agent.tools.base import Tool from nanobot.agent.tools.context import ContextAware, RequestContext from nanobot.agent.tools.runtime_state import RuntimeState -from nanobot.config_base import Base +from nanobot.config.tool_configs import MyToolConfig if TYPE_CHECKING: from nanobot.agent.subagent import SubagentStatus -class MyToolConfig(Base): - """Self-inspection tool configuration.""" - enable: bool = True - 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): diff --git a/nanobot/agent/tools/shell.py b/nanobot/agent/tools/shell.py index 808beaf09..62f794ee9 100644 --- a/nanobot/agent/tools/shell.py +++ b/nanobot/agent/tools/shell.py @@ -13,7 +13,6 @@ from pathlib import Path from typing import Any from loguru import logger -from pydantic import Field from nanobot.agent.tools.base import Tool, tool_parameters from nanobot.agent.tools.context import current_request_session_key @@ -34,7 +33,7 @@ from nanobot.agent.tools.schema import ( tool_parameters_schema, ) from nanobot.config.paths import get_media_dir -from nanobot.config_base import Base +from nanobot.config.tool_configs import ExecToolConfig from nanobot.security.workspace_access import current_scope_allows_loopback, current_tool_workspace from nanobot.security.workspace_policy import is_path_within @@ -51,18 +50,6 @@ _WORKSPACE_BOUNDARY_NOTE = ( ) -class ExecToolConfig(Base): - """Shell exec tool configuration.""" - enable: bool = True - timeout: int = Field(default=60, ge=0) # Hard timeout (s); 0 = no limit. Not capped by the per-call max. - path_prepend: str = "" - path_append: str = "" - sandbox: str = "" - allowed_env_keys: list[str] = Field(default_factory=list) - allow_patterns: list[str] = Field(default_factory=list) - deny_patterns: list[str] = Field(default_factory=list) - - @dataclass(slots=True) class _PreparedCommand: command: str diff --git a/nanobot/agent/tools/web.py b/nanobot/agent/tools/web.py index 0d2302b51..e66f86bc1 100644 --- a/nanobot/agent/tools/web.py +++ b/nanobot/agent/tools/web.py @@ -12,7 +12,6 @@ from urllib.parse import quote, urljoin, urlparse import httpx from loguru import logger -from pydantic import Field from nanobot.agent.tools.base import Tool, tool_parameters from nanobot.agent.tools.schema import ( @@ -21,7 +20,7 @@ from nanobot.agent.tools.schema import ( StringSchema, tool_parameters_schema, ) -from nanobot.config_base import Base +from nanobot.config.tool_configs import WebFetchConfig, WebSearchConfig, WebToolsConfig from nanobot.utils.helpers import build_image_content_blocks # Shared constants @@ -35,29 +34,6 @@ _VOLCENGINE_TIME_RANGES = {"OneDay", "OneWeek", "OneMonth", "OneYear"} _VOLCENGINE_DATE_RANGE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}\.\.\d{4}-\d{2}-\d{2}$") -class WebSearchConfig(Base): - """Web search configuration.""" - provider: str = "duckduckgo" - api_key: str = "" - base_url: str = "" - max_results: int = 5 - timeout: int = 30 - - -class WebFetchConfig(Base): - """Web fetch tool configuration.""" - use_jina_reader: bool = True - - -class WebToolsConfig(Base): - """Web tools configuration.""" - enable: bool = True - proxy: str | None = None - user_agent: str | None = None - search: WebSearchConfig = Field(default_factory=WebSearchConfig) - fetch: WebFetchConfig = Field(default_factory=WebFetchConfig) - - def _strip_tags(text: str) -> str: """Remove HTML tags and decode entities.""" text = re.sub(r'', '', text, flags=re.I) diff --git a/nanobot/agent/turn_session.py b/nanobot/agent/turn_session.py new file mode 100644 index 000000000..da3c2a09c --- /dev/null +++ b/nanobot/agent/turn_session.py @@ -0,0 +1,115 @@ +"""Turn-scoped session metadata coordination.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from nanobot.session.manager import Session, SessionManager + + +class TurnSessionCoordinator: + """Owns in-flight turn metadata stored on a session.""" + + RUNTIME_CHECKPOINT_KEY = "runtime_checkpoint" + PENDING_USER_TURN_KEY = "pending_user_turn" + + def __init__(self, sessions: SessionManager | None) -> None: + self._sessions = sessions + + def set_runtime_checkpoint(self, session: Session, payload: dict[str, Any]) -> None: + """Persist the latest in-flight turn state into session metadata.""" + session.metadata[self.RUNTIME_CHECKPOINT_KEY] = payload + if self._sessions is not None: + self._sessions.save(session) + + def mark_pending_user_turn(self, session: Session) -> None: + session.metadata[self.PENDING_USER_TURN_KEY] = True + + def clear_pending_user_turn(self, session: Session) -> None: + session.metadata.pop(self.PENDING_USER_TURN_KEY, None) + + def clear_runtime_checkpoint(self, session: Session) -> None: + session.metadata.pop(self.RUNTIME_CHECKPOINT_KEY, None) + + @staticmethod + def checkpoint_message_key(message: dict[str, Any]) -> tuple[Any, ...]: + return ( + message.get("role"), + message.get("content"), + message.get("tool_call_id"), + message.get("name"), + message.get("tool_calls"), + message.get("reasoning_content"), + message.get("thinking_blocks"), + ) + + def restore_runtime_checkpoint(self, session: Session) -> bool: + """Materialize an unfinished turn into session history before a new request.""" + checkpoint = session.metadata.get(self.RUNTIME_CHECKPOINT_KEY) + if not isinstance(checkpoint, dict): + return False + + assistant_message = checkpoint.get("assistant_message") + completed_tool_results = checkpoint.get("completed_tool_results") or [] + pending_tool_calls = checkpoint.get("pending_tool_calls") or [] + + restored_messages: list[dict[str, Any]] = [] + if isinstance(assistant_message, dict): + restored = dict(assistant_message) + restored.setdefault("timestamp", datetime.now().isoformat()) + restored_messages.append(restored) + for message in completed_tool_results: + if isinstance(message, dict): + restored = dict(message) + restored.setdefault("timestamp", datetime.now().isoformat()) + restored_messages.append(restored) + for tool_call in pending_tool_calls: + if not isinstance(tool_call, dict): + continue + tool_id = tool_call.get("id") + name = ((tool_call.get("function") or {}).get("name")) or "tool" + restored_messages.append( + { + "role": "tool", + "tool_call_id": tool_id, + "name": name, + "content": "Error: Task interrupted before this tool finished.", + "timestamp": datetime.now().isoformat(), + } + ) + + overlap = 0 + max_overlap = min(len(session.messages), len(restored_messages)) + for size in range(max_overlap, 0, -1): + existing = session.messages[-size:] + restored = restored_messages[:size] + if all( + self.checkpoint_message_key(left) == self.checkpoint_message_key(right) + for left, right in zip(existing, restored) + ): + overlap = size + break + session.messages.extend(restored_messages[overlap:]) + + self.clear_pending_user_turn(session) + self.clear_runtime_checkpoint(session) + return True + + def restore_pending_user_turn(self, session: Session) -> bool: + """Close a turn that only persisted the user message before crashing.""" + if not session.metadata.get(self.PENDING_USER_TURN_KEY): + return False + + if session.messages and session.messages[-1].get("role") == "user": + session.messages.append( + { + "role": "assistant", + "content": "Error: Task interrupted before a response was generated.", + "timestamp": datetime.now().isoformat(), + } + ) + session.updated_at = datetime.now() + + self.clear_pending_user_turn(session) + return True diff --git a/nanobot/config/loader.py b/nanobot/config/loader.py index 0fd1aa4c5..697b57bf4 100644 --- a/nanobot/config/loader.py +++ b/nanobot/config/loader.py @@ -9,11 +9,10 @@ from typing import Any import pydantic from pydantic import BaseModel -from nanobot.config.schema import Config, _resolve_tool_config_refs +from nanobot.config.schema import Config # Global variable to store current config path (for multi-instance support) _current_config_path: Path | None = None -_schema_refs_ready = False def set_config_path(path: Path) -> None: @@ -39,11 +38,6 @@ def load_config(config_path: Path | None = None) -> Config: Returns: Loaded configuration object. """ - global _schema_refs_ready - if not _schema_refs_ready: - _resolve_tool_config_refs() - _schema_refs_ready = True - path = config_path or get_config_path() config = Config() diff --git a/nanobot/config/schema.py b/nanobot/config/schema.py index bc0b13491..c45b4833e 100644 --- a/nanobot/config/schema.py +++ b/nanobot/config/schema.py @@ -2,21 +2,25 @@ from __future__ import annotations from pathlib import Path -from typing import TYPE_CHECKING, Any, Literal +from typing import Any, Literal from pydantic import AliasChoices, ConfigDict, Field, model_validator from pydantic_settings import BaseSettings +import nanobot.config.tool_configs as _tool_configs from nanobot.config_base import Base from nanobot.cron.types import CronSchedule -if TYPE_CHECKING: - from nanobot.agent.tools.cli_apps import CliAppsToolConfig - from nanobot.agent.tools.filesystem import FileToolsConfig - from nanobot.agent.tools.image_generation import ImageGenerationToolConfig - from nanobot.agent.tools.self import MyToolConfig - from nanobot.agent.tools.shell import ExecToolConfig - from nanobot.agent.tools.web import WebToolsConfig +CliAppsToolConfig = _tool_configs.CliAppsToolConfig +ExecToolConfig = _tool_configs.ExecToolConfig +FileToolsConfig = _tool_configs.FileToolsConfig +ImageGenerationToolConfig = _tool_configs.ImageGenerationToolConfig +MCPServerConfig = _tool_configs.MCPServerConfig +MyToolConfig = _tool_configs.MyToolConfig +ToolsConfig = _tool_configs.ToolsConfig +WebFetchConfig = _tool_configs.WebFetchConfig +WebSearchConfig = _tool_configs.WebSearchConfig +WebToolsConfig = _tool_configs.WebToolsConfig class ChannelsConfig(Base): @@ -290,57 +294,6 @@ class GatewayConfig(Base): heartbeat: HeartbeatConfig = Field(default_factory=HeartbeatConfig) -class MCPServerConfig(Base): - """MCP server connection configuration (stdio or HTTP).""" - - type: Literal["stdio", "sse", "streamableHttp"] | None = None # auto-detected if omitted - command: str = "" # Stdio: command to run (e.g. "npx") - args: list[str] = Field(default_factory=list) # Stdio: command arguments - env: dict[str, str] = Field(default_factory=dict) # Stdio: extra env vars - cwd: str = "" # Stdio: working directory for MCP server runtime artifacts - url: str = "" # HTTP/SSE: endpoint URL - headers: dict[str, str] = Field(default_factory=dict) # HTTP/SSE: custom headers - tool_timeout: int = 30 # seconds before a tool call is cancelled - enabled_tools: list[str] = Field(default_factory=lambda: ["*"]) # Only register these tools; accepts raw MCP names or wrapped mcp__ names; ["*"] = all tools; [] = no tools - - -def _lazy_default(module_path: str, class_name: str) -> Any: - """Deferred import helper for ToolsConfig default factories.""" - import importlib - module = importlib.import_module(module_path) - return getattr(module, class_name)() - - -class ToolsConfig(Base): - """Tools configuration. - - Field types for tool-specific sub-configs are resolved via model_rebuild() - at the bottom of this file so tool config classes can stay next to their - tool implementations. - """ - - web: WebToolsConfig = Field(default_factory=lambda: _lazy_default("nanobot.agent.tools.web", "WebToolsConfig")) - exec: ExecToolConfig = Field(default_factory=lambda: _lazy_default("nanobot.agent.tools.shell", "ExecToolConfig")) - file: FileToolsConfig = Field(default_factory=lambda: _lazy_default("nanobot.agent.tools.filesystem", "FileToolsConfig")) - cli_apps: CliAppsToolConfig = Field(default_factory=lambda: _lazy_default("nanobot.agent.tools.cli_apps", "CliAppsToolConfig")) - my: MyToolConfig = Field(default_factory=lambda: _lazy_default("nanobot.agent.tools.self", "MyToolConfig")) - image_generation: ImageGenerationToolConfig = Field( - default_factory=lambda: _lazy_default("nanobot.agent.tools.image_generation", "ImageGenerationToolConfig"), - ) - restrict_to_workspace: bool = False # policy intent: keep tool access inside workspace when possible - webui_allow_local_service_access: bool = Field( - default=True, - validation_alias=AliasChoices( - "webuiAllowLocalServiceAccess", - "webui_allow_local_service_access", - "allowLocalPreviewAccess", - "allow_local_preview_access", - ), - ) # allow WebUI Full Access shell checks against localhost services; legacy allowLocalPreviewAccess still reads - mcp_servers: dict[str, MCPServerConfig] = Field(default_factory=dict) - ssrf_whitelist: list[str] = Field(default_factory=list) # CIDR ranges to exempt from SSRF blocking (e.g. ["100.64.0.0/10"] for Tailscale) - - class Config(BaseSettings): """Root configuration for nanobot.""" @@ -356,11 +309,6 @@ class Config(BaseSettings): validation_alias=AliasChoices("modelPresets", "model_presets"), ) - def __init__(self, **values: Any) -> None: - if not type(self).__pydantic_complete__: - _resolve_tool_config_refs() - super().__init__(**values) - @model_validator(mode="after") def _validate_model_preset(self) -> "Config": if "default" in self.model_presets: @@ -548,43 +496,3 @@ class Config(BaseSettings): return None model_config = ConfigDict(env_prefix="NANOBOT_", env_nested_delimiter="__") - - -def _resolve_tool_config_refs() -> None: - """Resolve forward references in ToolsConfig by importing tool config classes. - - Must be called after all modules are loaded (breaks circular imports). - Re-exports the classes into this module's namespace so existing imports - like ``from nanobot.config.schema import ExecToolConfig`` continue to work. - """ - import sys - - from nanobot.agent.tools.cli_apps import CliAppsToolConfig - from nanobot.agent.tools.filesystem import FileToolsConfig - from nanobot.agent.tools.image_generation import ImageGenerationToolConfig - from nanobot.agent.tools.self import MyToolConfig - from nanobot.agent.tools.shell import ExecToolConfig - from nanobot.agent.tools.web import WebFetchConfig, WebSearchConfig, WebToolsConfig - - # Re-export into this module's namespace - mod = sys.modules[__name__] - mod.ExecToolConfig = ExecToolConfig # type: ignore[attr-defined] - mod.FileToolsConfig = FileToolsConfig # type: ignore[attr-defined] - mod.CliAppsToolConfig = CliAppsToolConfig # type: ignore[attr-defined] - mod.WebToolsConfig = WebToolsConfig # type: ignore[attr-defined] - mod.WebSearchConfig = WebSearchConfig # type: ignore[attr-defined] - mod.WebFetchConfig = WebFetchConfig # type: ignore[attr-defined] - mod.MyToolConfig = MyToolConfig # type: ignore[attr-defined] - mod.ImageGenerationToolConfig = ImageGenerationToolConfig # type: ignore[attr-defined] - - ToolsConfig.model_rebuild() - Config.model_rebuild() - - -# Eagerly resolve when the import chain allows it (no circular deps at this -# point). If it fails (first import triggers a cycle), the rebuild will -# happen lazily when Config/ToolsConfig is first used at runtime. -try: - _resolve_tool_config_refs() -except ImportError: - pass diff --git a/nanobot/config/tool_configs.py b/nanobot/config/tool_configs.py new file mode 100644 index 000000000..ea8b7c5e2 --- /dev/null +++ b/nanobot/config/tool_configs.py @@ -0,0 +1,119 @@ +"""Tool configuration models with no tool-runtime imports.""" + +from __future__ import annotations + +from typing import Literal + +from pydantic import AliasChoices, Field + +from nanobot.config_base import Base + + +class WebSearchConfig(Base): + """Web search configuration.""" + + provider: str = "duckduckgo" + api_key: str = "" + base_url: str = "" + max_results: int = 5 + timeout: int = 30 + + +class WebFetchConfig(Base): + """Web fetch tool configuration.""" + + use_jina_reader: bool = True + + +class WebToolsConfig(Base): + """Web tools configuration.""" + + enable: bool = True + proxy: str | None = None + user_agent: str | None = None + search: WebSearchConfig = Field(default_factory=WebSearchConfig) + fetch: WebFetchConfig = Field(default_factory=WebFetchConfig) + + +class ExecToolConfig(Base): + """Shell exec tool configuration.""" + + enable: bool = True + timeout: int = Field(default=60, ge=0) + path_prepend: str = "" + path_append: str = "" + sandbox: str = "" + allowed_env_keys: list[str] = Field(default_factory=list) + allow_patterns: list[str] = Field(default_factory=list) + deny_patterns: list[str] = Field(default_factory=list) + + +class FileToolsConfig(Base): + """Filesystem tools configuration.""" + + enable: bool = True + + +class CliAppsToolConfig(Base): + """CLI Apps tool configuration.""" + + enable: bool = True + install_timeout: int = Field(default=300, ge=1, le=3600) + run_timeout: int = Field(default=60, ge=1, le=600) + catalog_ttl_seconds: int = Field(default=3600, ge=60, le=86_400) + + +class MyToolConfig(Base): + """Self-inspection tool configuration.""" + + enable: bool = True + allow_set: bool = False + + +class ImageGenerationToolConfig(Base): + """Image generation tool configuration.""" + + enabled: bool = False + provider: str = "openrouter" + model: str = "openai/gpt-5.4-image-2" + default_aspect_ratio: str = "1:1" + default_image_size: str = "1K" + max_images_per_turn: int = Field(default=4, ge=1, le=8) + save_dir: str = "generated" + + +class MCPServerConfig(Base): + """MCP server connection configuration (stdio or HTTP).""" + + type: Literal["stdio", "sse", "streamableHttp"] | None = None + command: str = "" + args: list[str] = Field(default_factory=list) + env: dict[str, str] = Field(default_factory=dict) + cwd: str = "" + url: str = "" + headers: dict[str, str] = Field(default_factory=dict) + tool_timeout: int = 30 + enabled_tools: list[str] = Field(default_factory=lambda: ["*"]) + + +class ToolsConfig(Base): + """Tools configuration.""" + + web: WebToolsConfig = Field(default_factory=WebToolsConfig) + exec: ExecToolConfig = Field(default_factory=ExecToolConfig) + file: FileToolsConfig = Field(default_factory=FileToolsConfig) + cli_apps: CliAppsToolConfig = Field(default_factory=CliAppsToolConfig) + my: MyToolConfig = Field(default_factory=MyToolConfig) + image_generation: ImageGenerationToolConfig = Field(default_factory=ImageGenerationToolConfig) + restrict_to_workspace: bool = False + webui_allow_local_service_access: bool = Field( + default=True, + validation_alias=AliasChoices( + "webuiAllowLocalServiceAccess", + "webui_allow_local_service_access", + "allowLocalPreviewAccess", + "allow_local_preview_access", + ), + ) + mcp_servers: dict[str, MCPServerConfig] = Field(default_factory=dict) + ssrf_whitelist: list[str] = Field(default_factory=list)