refactor agent config and loop boundaries

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 self.runtime_models.llm_runtime()
-        return LLMRuntime(self.provider, self.model)
 
-    _RUNTIME_CHECKPOINT_KEY = "runtime_checkpoint"
+    _RUNTIME_CHECKPOINT_KEY = TurnSessionCoordinator.RUNTIME_CHECKPOINT_KEY
-    _PENDING_USER_TURN_KEY = "pending_user_turn"
+    _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
+        self.runtime_models.apply_provider_snapshot(
-        model = snapshot.model
+            snapshot,
-        context_window_tokens = snapshot.context_window_tokens
+            publish_update=publish_update,
-        old_model = self.model
+            model_preset=model_preset,
-        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)
 
     def _refresh_provider_snapshot(self) -> None:
-        if self._provider_snapshot_loader is None:
+        self.runtime_models.refresh_provider_snapshot()
-            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)
 
     @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(
+        return self.runtime_models.build_model_preset_snapshot(name)
-            name=name,
-            presets=self.model_presets,
-            provider=self.provider,
-            loader=self._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."""
-        name = preset_helpers.normalize_preset_name(name, self.model_presets)
+        self.runtime_models.set_model_preset(name, publish_update=publish_update)
-        snapshot = self._build_model_preset_snapshot(name)
-        self._apply_provider_snapshot(snapshot, publish_update=publish_update, model_preset=name)
-        self._active_preset = name
 
     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.
+                        callbacks = self.streaming.build_callbacks(msg)
-                        stream_base_id = f"{msg.session_key}:{time.time_ns()}"
+                        on_stream = callbacks.on_stream
-                        stream_segment = 0
+                        on_stream_end = callbacks.on_stream_end
-
-                        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
 
                     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._turn_session_coordinator().set_runtime_checkpoint(session, payload)
-        self.sessions.save(session)
+
+    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:
+        self._turn_session_coordinator().clear_runtime_checkpoint(session)
-            session.metadata.pop(self._RUNTIME_CHECKPOINT_KEY, None)
 
     @staticmethod
     def _checkpoint_message_key(message: dict[str, Any]) -> tuple[Any, ...]:
-        return (
+        return TurnSessionCoordinator.checkpoint_message_key(message)
-            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."""
-        from datetime import datetime
+        return self._turn_session_coordinator().restore_runtime_checkpoint(session)
-
-        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."""
-        from datetime import datetime
+        return self._turn_session_coordinator().restore_pending_user_turn(session)
-
-        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
 
     async def process_direct(
         self,

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

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)

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.agent.tools.schema import (
-from nanobot.security.workspace_access import current_tool_workspace
+    ArraySchema,
+    BooleanSchema,
+    IntegerSchema,
+    StringSchema,
+    tool_parameters_schema,
+)
 from nanobot.apps.cli import CliAppError, CliAppManager, CliAppsRuntimeConfig
-from nanobot.config_base import Base
+from nanobot.config.tool_configs import CliAppsToolConfig
-
+from nanobot.security.workspace_access import current_tool_workspace
-
-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)
 
 
 @tool_parameters(

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."""
 

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(

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):

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

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'<script[\s\S]*?</script>', '', text, flags=re.I)

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

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()

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:
+CliAppsToolConfig = _tool_configs.CliAppsToolConfig
-    from nanobot.agent.tools.cli_apps import CliAppsToolConfig
+ExecToolConfig = _tool_configs.ExecToolConfig
-    from nanobot.agent.tools.filesystem import FileToolsConfig
+FileToolsConfig = _tool_configs.FileToolsConfig
-    from nanobot.agent.tools.image_generation import ImageGenerationToolConfig
+ImageGenerationToolConfig = _tool_configs.ImageGenerationToolConfig
-    from nanobot.agent.tools.self import MyToolConfig
+MCPServerConfig = _tool_configs.MCPServerConfig
-    from nanobot.agent.tools.shell import ExecToolConfig
+MyToolConfig = _tool_configs.MyToolConfig
-    from nanobot.agent.tools.web import WebToolsConfig
+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_<server>_<tool> 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

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)