refactor: break up capture_error() into testable helpers

Extract 5 focused helpers from the 138-line capture_error():
- _extract_origin(): walk traceback for file/line
- _log_error_event(): log to event log (best-effort)
- _create_bug_report(): create task and log creation event
- _send_error_notification(): push notification
- _record_to_session(): forward to session recorder

capture_error() now orchestrates the helpers in ~25 lines.
Added tests for each new helper.

Fixes #506

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/config.py

@@ -10,6 +10,11 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
 APP_START_TIME: _datetime = _datetime.now(UTC)
 
 
+def normalize_ollama_url(url: str) -> str:
+    """Replace localhost with 127.0.0.1 to avoid IPv6 resolution delays."""
+    return url.replace("localhost", "127.0.0.1")
+
+
 class Settings(BaseSettings):
     """Central configuration — all env-var access goes through this class."""
 
@@ -19,6 +24,11 @@ class Settings(BaseSettings):
     # Ollama host — override with OLLAMA_URL env var or .env file
     ollama_url: str = "http://localhost:11434"
 
+    @property
+    def normalized_ollama_url(self) -> str:
+        """Return ollama_url with localhost replaced by 127.0.0.1."""
+        return normalize_ollama_url(self.ollama_url)
+
     # LLM model passed to Agno/Ollama — override with OLLAMA_MODEL
     # qwen3:30b is the primary model — better reasoning and tool calling
     # than llama3.1:8b-instruct while still running locally on modest hardware.
@@ -392,7 +402,7 @@ def check_ollama_model_available(model_name: str) -> bool:
         import json
         import urllib.request
 
-        url = settings.ollama_url.replace("localhost", "127.0.0.1")
+        url = settings.normalized_ollama_url
         req = urllib.request.Request(
             f"{url}/api/tags",
             method="GET",

src/dashboard/app.py

@@ -329,33 +329,35 @@ async def _discord_token_watcher() -> None:
                 logger.warning("Discord auto-start failed: %s", exc)
 
 
-@asynccontextmanager
+def _startup_init() -> None:
-async def lifespan(app: FastAPI):
+    """Validate config and enable event persistence."""
-    """Application lifespan manager with non-blocking startup."""
-
-    # Validate security config (no-op in test mode)
     from config import validate_startup
 
     validate_startup()
 
-    # Enable event persistence (unified EventBus + swarm event_log)
     from infrastructure.events.bus import init_event_bus_persistence
 
     init_event_bus_persistence()
 
-    # Create all background tasks without waiting for them
-    briefing_task = asyncio.create_task(_briefing_scheduler())
-    thinking_task = asyncio.create_task(_thinking_scheduler())
-    loop_qa_task = asyncio.create_task(_loop_qa_scheduler())
-    presence_task = asyncio.create_task(_presence_watcher())
-
-    # Initialize Spark Intelligence engine
     from spark.engine import get_spark_engine
 
     if get_spark_engine().enabled:
         logger.info("Spark Intelligence active — event capture enabled")
 
-    # Auto-prune old vector store memories on startup
+
+def _startup_background_tasks() -> list[asyncio.Task]:
+    """Spawn all recurring background tasks (non-blocking)."""
+    return [
+        asyncio.create_task(_briefing_scheduler()),
+        asyncio.create_task(_thinking_scheduler()),
+        asyncio.create_task(_loop_qa_scheduler()),
+        asyncio.create_task(_presence_watcher()),
+        asyncio.create_task(_start_chat_integrations_background()),
+    ]
+
+
+def _startup_pruning() -> None:
+    """Auto-prune old memories, thoughts, and events on startup."""
     if settings.memory_prune_days > 0:
         try:
             from timmy.memory_system import prune_memories
@@ -373,7 +375,6 @@ async def lifespan(app: FastAPI):
         except Exception as exc:
             logger.debug("Memory auto-prune skipped: %s", exc)
 
-    # Auto-prune old thoughts on startup
     if settings.thoughts_prune_days > 0:
         try:
             from timmy.thinking import thinking_engine
@@ -391,7 +392,6 @@ async def lifespan(app: FastAPI):
         except Exception as exc:
             logger.debug("Thought auto-prune skipped: %s", exc)
 
-    # Auto-prune old system events on startup
     if settings.events_prune_days > 0:
         try:
             from swarm.event_log import prune_old_events
@@ -409,7 +409,6 @@ async def lifespan(app: FastAPI):
         except Exception as exc:
             logger.debug("Event auto-prune skipped: %s", exc)
 
-    # Warn if memory vault exceeds size limit
     if settings.memory_vault_max_mb > 0:
         try:
             vault_path = Path(settings.repo_root) / "memory" / "notes"
@@ -425,6 +424,42 @@ async def lifespan(app: FastAPI):
         except Exception as exc:
             logger.debug("Vault size check skipped: %s", exc)
 
+
+async def _shutdown_cleanup(
+    bg_tasks: list[asyncio.Task],
+    workshop_heartbeat,
+) -> None:
+    """Stop chat bots, MCP sessions, heartbeat, and cancel background tasks."""
+    from integrations.chat_bridge.vendors.discord import discord_bot
+    from integrations.telegram_bot.bot import telegram_bot
+
+    await discord_bot.stop()
+    await telegram_bot.stop()
+
+    try:
+        from timmy.mcp_tools import close_mcp_sessions
+
+        await close_mcp_sessions()
+    except Exception as exc:
+        logger.debug("MCP shutdown: %s", exc)
+
+    await workshop_heartbeat.stop()
+
+    for task in bg_tasks:
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Application lifespan manager with non-blocking startup."""
+    _startup_init()
+    bg_tasks = _startup_background_tasks()
+    _startup_pruning()
+
     # Start Workshop presence heartbeat with WS relay
     from dashboard.routes.world import broadcast_world_state
     from timmy.workshop_state import WorkshopHeartbeat
@@ -432,10 +467,7 @@ async def lifespan(app: FastAPI):
     workshop_heartbeat = WorkshopHeartbeat(on_change=broadcast_world_state)
     await workshop_heartbeat.start()
 
-    # Start chat integrations in background
+    # Register session logger with error capture
-    chat_task = asyncio.create_task(_start_chat_integrations_background())
-
-    # Register session logger with error capture (breaks infrastructure → timmy circular dep)
     try:
         from infrastructure.error_capture import register_error_recorder
         from timmy.session_logger import get_session_logger
@@ -448,30 +480,7 @@ async def lifespan(app: FastAPI):
 
     yield
 
-    # Cleanup on shutdown
+    await _shutdown_cleanup(bg_tasks, workshop_heartbeat)
-    from integrations.chat_bridge.vendors.discord import discord_bot
-    from integrations.telegram_bot.bot import telegram_bot
-
-    await discord_bot.stop()
-    await telegram_bot.stop()
-
-    # Close MCP tool server sessions
-    try:
-        from timmy.mcp_tools import close_mcp_sessions
-
-        await close_mcp_sessions()
-    except Exception as exc:
-        logger.debug("MCP shutdown: %s", exc)
-
-    await workshop_heartbeat.stop()
-
-    for task in [briefing_task, thinking_task, chat_task, loop_qa_task, presence_task]:
-        if task:
-            task.cancel()
-            try:
-                await task
-            except asyncio.CancelledError:
-                pass
 
 
 app = FastAPI(

src/dashboard/routes/health.py

@@ -65,7 +65,7 @@ def _check_ollama_sync() -> DependencyStatus:
     try:
         import urllib.request
 
-        url = settings.ollama_url.replace("localhost", "127.0.0.1")
+        url = settings.normalized_ollama_url
         req = urllib.request.Request(
             f"{url}/api/tags",
             method="GET",

src/infrastructure/error_capture.py

@@ -100,48 +100,25 @@ def _get_git_context() -> dict:
         return {"branch": "unknown", "commit": "unknown"}
 
 
-def capture_error(
+def _extract_origin(exc: Exception) -> tuple[str, int]:
-    exc: Exception,
+    """Walk the traceback to find the deepest file and line number."""
-    source: str = "unknown",
-    context: dict | None = None,
-) -> str | None:
-    """Capture an error and optionally create a bug report.
-
-    Args:
-        exc: The exception to capture
-        source: Module/component where the error occurred
-        context: Optional dict of extra context (request path, etc.)
-
-    Returns:
-        Task ID of the created bug report, or None if deduplicated/disabled
-    """
-    from config import settings
-
-    if not settings.error_feedback_enabled:
-        return None
-
-    error_hash = _stack_hash(exc)
-
-    if _is_duplicate(error_hash):
-        logger.debug("Duplicate error suppressed: %s (hash=%s)", exc, error_hash)
-        return None
-
-    # Format the stack trace
-    tb_str = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
-
-    # Extract file/line from traceback
     tb_obj = exc.__traceback__
-    affected_file = "unknown"
-    affected_line = 0
     while tb_obj and tb_obj.tb_next:
         tb_obj = tb_obj.tb_next
     if tb_obj:
-        affected_file = tb_obj.tb_frame.f_code.co_filename
+        return tb_obj.tb_frame.f_code.co_filename, tb_obj.tb_lineno
-        affected_line = tb_obj.tb_lineno
+    return "unknown", 0
 
-    git_ctx = _get_git_context()
 
-    # 1. Log to event_log
+def _log_error_event(
+    exc: Exception,
+    source: str,
+    error_hash: str,
+    affected_file: str,
+    affected_line: int,
+    git_ctx: dict,
+) -> None:
+    """Log the error to the event log (best-effort)."""
     try:
         from swarm.event_log import EventType, log_event
 
@@ -161,8 +138,18 @@ def capture_error(
     except Exception as log_exc:
         logger.debug("Failed to log error event: %s", log_exc)
 
-    # 2. Create bug report task
+
-    task_id = None
+def _create_bug_report(
+    exc: Exception,
+    source: str,
+    error_hash: str,
+    affected_file: str,
+    affected_line: int,
+    git_ctx: dict,
+    tb_str: str,
+    context: dict | None,
+) -> str | None:
+    """Create a bug report task and return its ID (best-effort)."""
     try:
         from swarm.task_queue.models import create_task
 
@@ -193,29 +180,30 @@ def capture_error(
             auto_approve=True,
             task_type="bug_report",
         )
-        task_id = task.id
 
-        # Log the creation event
         try:
             from swarm.event_log import EventType, log_event
 
             log_event(
                 EventType.BUG_REPORT_CREATED,
                 source=source,
-                task_id=task_id,
+                task_id=task.id,
                 data={
                     "error_hash": error_hash,
                     "title": title[:100],
                 },
             )
-        except Exception as exc:
+        except Exception as log_exc:
-            logger.warning("Bug report screenshot error: %s", exc)
+            logger.warning("Bug report log error: %s", log_exc)
-            pass
 
+        return task.id
     except Exception as task_exc:
         logger.debug("Failed to create bug report task: %s", task_exc)
+        return None
 
-    # 3. Send notification
+
+def _send_error_notification(exc: Exception, source: str) -> None:
+    """Push a notification about the captured error (best-effort)."""
     try:
         from infrastructure.notifications.push import notifier
 
@@ -224,11 +212,12 @@ def capture_error(
             message=f"{type(exc).__name__} in {source}: {str(exc)[:80]}",
             category="system",
         )
-    except Exception as exc:
+    except Exception as notify_exc:
-        logger.warning("Bug report notification error: %s", exc)
+        logger.warning("Bug report notification error: %s", notify_exc)
-        pass
 
-    # 4. Record in session logger (via registered callback)
+
+def _record_to_session(exc: Exception, source: str) -> None:
+    """Forward the error to the registered session recorder (best-effort)."""
     if _error_recorder is not None:
         try:
             _error_recorder(
@@ -238,4 +227,44 @@ def capture_error(
         except Exception as log_exc:
             logger.warning("Bug report session logging error: %s", log_exc)
 
+
+def capture_error(
+    exc: Exception,
+    source: str = "unknown",
+    context: dict | None = None,
+) -> str | None:
+    """Capture an error and optionally create a bug report.
+
+    Args:
+        exc: The exception to capture
+        source: Module/component where the error occurred
+        context: Optional dict of extra context (request path, etc.)
+
+    Returns:
+        Task ID of the created bug report, or None if deduplicated/disabled
+    """
+    from config import settings
+
+    if not settings.error_feedback_enabled:
+        return None
+
+    error_hash = _stack_hash(exc)
+
+    if _is_duplicate(error_hash):
+        logger.debug("Duplicate error suppressed: %s (hash=%s)", exc, error_hash)
+        return None
+
+    tb_str = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
+    affected_file, affected_line = _extract_origin(exc)
+    git_ctx = _get_git_context()
+
+    _log_error_event(exc, source, error_hash, affected_file, affected_line, git_ctx)
+
+    task_id = _create_bug_report(
+        exc, source, error_hash, affected_file, affected_line, git_ctx, tb_str, context
+    )
+
+    _send_error_notification(exc, source)
+    _record_to_session(exc, source)
+
     return task_id

src/infrastructure/models/multimodal.py

@@ -13,7 +13,7 @@ import logging
 from dataclasses import dataclass, field
 from enum import Enum, auto
 
-from config import settings
+from config import normalize_ollama_url, settings
 
 logger = logging.getLogger(__name__)
 
@@ -307,7 +307,7 @@ class MultiModalManager:
             import json
             import urllib.request
 
-            url = self.ollama_url.replace("localhost", "127.0.0.1")
+            url = normalize_ollama_url(self.ollama_url)
             req = urllib.request.Request(
                 f"{url}/api/tags",
                 method="GET",
@@ -462,7 +462,7 @@ class MultiModalManager:
 
             logger.info("Pulling model: %s", model_name)
 
-            url = self.ollama_url.replace("localhost", "127.0.0.1")
+            url = normalize_ollama_url(self.ollama_url)
             req = urllib.request.Request(
                 f"{url}/api/pull",
                 method="POST",

src/infrastructure/router/cascade.py

@@ -388,6 +388,101 @@ class CascadeRouter:
 
         return None
 
+    def _select_model(
+        self, provider: Provider, model: str | None, content_type: ContentType
+    ) -> tuple[str | None, bool]:
+        """Select the best model for the request, with vision fallback.
+
+        Returns:
+            Tuple of (selected_model, is_fallback_model).
+        """
+        selected_model = model or provider.get_default_model()
+        is_fallback = False
+
+        if content_type != ContentType.TEXT and selected_model:
+            if provider.type == "ollama" and self._mm_manager:
+                from infrastructure.models.multimodal import ModelCapability
+
+                if content_type == ContentType.VISION:
+                    supports = self._mm_manager.model_supports(
+                        selected_model, ModelCapability.VISION
+                    )
+                    if not supports:
+                        fallback = self._get_fallback_model(provider, selected_model, content_type)
+                        if fallback:
+                            logger.info(
+                                "Model %s doesn't support vision, falling back to %s",
+                                selected_model,
+                                fallback,
+                            )
+                            selected_model = fallback
+                            is_fallback = True
+                        else:
+                            logger.warning(
+                                "No vision-capable model found on %s, trying anyway",
+                                provider.name,
+                            )
+
+        return selected_model, is_fallback
+
+    async def _attempt_with_retry(
+        self,
+        provider: Provider,
+        messages: list[dict],
+        model: str | None,
+        temperature: float,
+        max_tokens: int | None,
+        content_type: ContentType,
+    ) -> dict:
+        """Try a provider with retries, returning the result dict.
+
+        Raises:
+            RuntimeError: If all retry attempts fail.
+            Returns error strings collected during retries via the exception message.
+        """
+        errors: list[str] = []
+        for attempt in range(self.config.max_retries_per_provider):
+            try:
+                return await self._try_provider(
+                    provider=provider,
+                    messages=messages,
+                    model=model,
+                    temperature=temperature,
+                    max_tokens=max_tokens,
+                    content_type=content_type,
+                )
+            except Exception as exc:
+                error_msg = str(exc)
+                logger.warning(
+                    "Provider %s attempt %d failed: %s",
+                    provider.name,
+                    attempt + 1,
+                    error_msg,
+                )
+                errors.append(f"{provider.name}: {error_msg}")
+
+                if attempt < self.config.max_retries_per_provider - 1:
+                    await asyncio.sleep(self.config.retry_delay_seconds)
+
+        raise RuntimeError("; ".join(errors))
+
+    def _is_provider_available(self, provider: Provider) -> bool:
+        """Check if a provider should be tried (enabled + circuit breaker)."""
+        if not provider.enabled:
+            logger.debug("Skipping %s (disabled)", provider.name)
+            return False
+
+        if provider.status == ProviderStatus.UNHEALTHY:
+            if self._can_close_circuit(provider):
+                provider.circuit_state = CircuitState.HALF_OPEN
+                provider.half_open_calls = 0
+                logger.info("Circuit breaker half-open for %s", provider.name)
+            else:
+                logger.debug("Skipping %s (circuit open)", provider.name)
+                return False
+
+        return True
+
     async def complete(
         self,
         messages: list[dict],
@@ -414,7 +509,6 @@ class CascadeRouter:
         Raises:
             RuntimeError: If all providers fail
         """
-        # Detect content type for multi-modal routing
         content_type = self._detect_content_type(messages)
         if content_type != ContentType.TEXT:
             logger.debug("Detected %s content, selecting appropriate model", content_type.value)
@@ -422,93 +516,34 @@ class CascadeRouter:
         errors = []
 
         for provider in self.providers:
-            # Skip disabled providers
+            if not self._is_provider_available(provider):
-            if not provider.enabled:
-                logger.debug("Skipping %s (disabled)", provider.name)
                 continue
 
-            # Skip unhealthy providers (circuit breaker)
+            selected_model, is_fallback_model = self._select_model(provider, model, content_type)
-            if provider.status == ProviderStatus.UNHEALTHY:
-                # Check if circuit breaker can close
-                if self._can_close_circuit(provider):
-                    provider.circuit_state = CircuitState.HALF_OPEN
-                    provider.half_open_calls = 0
-                    logger.info("Circuit breaker half-open for %s", provider.name)
-                else:
-                    logger.debug("Skipping %s (circuit open)", provider.name)
-                    continue
 
-            # Determine which model to use
+            try:
-            selected_model = model or provider.get_default_model()
+                result = await self._attempt_with_retry(
-            is_fallback_model = False
+                    provider,
+                    messages,
+                    selected_model,
+                    temperature,
+                    max_tokens,
+                    content_type,
+                )
+            except RuntimeError as exc:
+                errors.append(str(exc))
+                self._record_failure(provider)
+                continue
 
-            # For non-text content, check if model supports it
+            self._record_success(provider, result.get("latency_ms", 0))
-            if content_type != ContentType.TEXT and selected_model:
+            return {
-                if provider.type == "ollama" and self._mm_manager:
+                "content": result["content"],
-                    from infrastructure.models.multimodal import ModelCapability
+                "provider": provider.name,
+                "model": result.get("model", selected_model or provider.get_default_model()),
+                "latency_ms": result.get("latency_ms", 0),
+                "is_fallback_model": is_fallback_model,
+            }
 
-                    # Check if selected model supports the required capability
-                    if content_type == ContentType.VISION:
-                        supports = self._mm_manager.model_supports(
-                            selected_model, ModelCapability.VISION
-                        )
-                        if not supports:
-                            # Find fallback model
-                            fallback = self._get_fallback_model(
-                                provider, selected_model, content_type
-                            )
-                            if fallback:
-                                logger.info(
-                                    "Model %s doesn't support vision, falling back to %s",
-                                    selected_model,
-                                    fallback,
-                                )
-                                selected_model = fallback
-                                is_fallback_model = True
-                            else:
-                                logger.warning(
-                                    "No vision-capable model found on %s, trying anyway",
-                                    provider.name,
-                                )
-
-            # Try this provider
-            for attempt in range(self.config.max_retries_per_provider):
-                try:
-                    result = await self._try_provider(
-                        provider=provider,
-                        messages=messages,
-                        model=selected_model,
-                        temperature=temperature,
-                        max_tokens=max_tokens,
-                        content_type=content_type,
-                    )
-
-                    # Success! Update metrics and return
-                    self._record_success(provider, result.get("latency_ms", 0))
-                    return {
-                        "content": result["content"],
-                        "provider": provider.name,
-                        "model": result.get(
-                            "model", selected_model or provider.get_default_model()
-                        ),
-                        "latency_ms": result.get("latency_ms", 0),
-                        "is_fallback_model": is_fallback_model,
-                    }
-
-                except Exception as exc:
-                    error_msg = str(exc)
-                    logger.warning(
-                        "Provider %s attempt %d failed: %s", provider.name, attempt + 1, error_msg
-                    )
-                    errors.append(f"{provider.name}: {error_msg}")
-
-                    if attempt < self.config.max_retries_per_provider - 1:
-                        await asyncio.sleep(self.config.retry_delay_seconds)
-
-            # All retries failed for this provider
-            self._record_failure(provider)
-
-        # All providers failed
         raise RuntimeError(f"All providers failed: {'; '.join(errors)}")
 
     async def _try_provider(

src/timmy/agent.py

@@ -63,7 +63,7 @@ def _pull_model(model_name: str) -> bool:
 
         logger.info("Pulling model: %s", model_name)
 
-        url = settings.ollama_url.replace("localhost", "127.0.0.1")
+        url = settings.normalized_ollama_url
         req = urllib.request.Request(
             f"{url}/api/pull",
             method="POST",
@@ -197,6 +197,90 @@ def _resolve_backend(requested: str | None) -> str:
     return "ollama"
 
 
+def _build_tools_list(use_tools: bool, skip_mcp: bool, model_name: str) -> list:
+    """Assemble the tools list based on model capability and MCP flags.
+
+    Returns a list of Toolkit / MCPTools objects, or an empty list.
+    """
+    if not use_tools:
+        logger.info("Tools disabled for model %s (too small for reliable tool calling)", model_name)
+        return []
+
+    tools_list: list = [create_full_toolkit()]
+
+    # Add MCP tool servers (lazy-connected on first arun()).
+    # Skipped when skip_mcp=True — MCP's stdio transport uses anyio cancel
+    # scopes that conflict with asyncio background task cancellation (#72).
+    if not skip_mcp:
+        try:
+            from timmy.mcp_tools import create_filesystem_mcp_tools, create_gitea_mcp_tools
+
+            gitea_mcp = create_gitea_mcp_tools()
+            if gitea_mcp:
+                tools_list.append(gitea_mcp)
+
+            fs_mcp = create_filesystem_mcp_tools()
+            if fs_mcp:
+                tools_list.append(fs_mcp)
+        except Exception as exc:
+            logger.debug("MCP tools unavailable: %s", exc)
+
+    return tools_list
+
+
+def _build_prompt(use_tools: bool, session_id: str) -> str:
+    """Build the full system prompt with optional memory context."""
+    base_prompt = get_system_prompt(tools_enabled=use_tools, session_id=session_id)
+
+    try:
+        from timmy.memory_system import memory_system
+
+        memory_context = memory_system.get_system_context()
+        if memory_context:
+            # Smaller budget for small models — expanded prompt uses more tokens
+            max_context = 2000 if not use_tools else 8000
+            if len(memory_context) > max_context:
+                memory_context = memory_context[:max_context] + "\n... [truncated]"
+            return (
+                f"{base_prompt}\n\n"
+                f"## GROUNDED CONTEXT (verified sources — cite when using)\n\n"
+                f"{memory_context}"
+            )
+    except Exception as exc:
+        logger.warning("Failed to load memory context: %s", exc)
+
+    return base_prompt
+
+
+def _create_ollama_agent(
+    *,
+    db_file: str,
+    model_name: str,
+    tools_list: list,
+    full_prompt: str,
+    use_tools: bool,
+) -> Agent:
+    """Construct the Agno Agent with Ollama backend and warm up the model."""
+    model_kwargs = {}
+    if settings.ollama_num_ctx > 0:
+        model_kwargs["options"] = {"num_ctx": settings.ollama_num_ctx}
+
+    agent = Agent(
+        name="Agent",
+        model=Ollama(id=model_name, host=settings.ollama_url, timeout=300, **model_kwargs),
+        db=SqliteDb(db_file=db_file),
+        description=full_prompt,
+        add_history_to_context=True,
+        num_history_runs=20,
+        markdown=False,
+        tools=tools_list if tools_list else None,
+        tool_call_limit=settings.max_agent_steps if use_tools else None,
+        telemetry=settings.telemetry_enabled,
+    )
+    _warmup_model(model_name)
+    return agent
+
+
 def create_timmy(
     db_file: str = "timmy.db",
     backend: str | None = None,
@@ -238,16 +322,12 @@ def create_timmy(
         return TimmyAirLLMAgent(model_size=size)
 
     # Default: Ollama via Agno.
-    # Resolve model with automatic pulling and fallback
     model_name, is_fallback = _resolve_model_with_fallback(
         requested_model=None,
         require_vision=False,
         auto_pull=True,
     )
 
-    # If Ollama is completely unreachable, fail loudly.
-    # Sovereignty: never silently send data to a cloud API.
-    # Use --backend claude explicitly if you want cloud inference.
     if not _check_model_available(model_name):
         logger.error(
             "Ollama unreachable and no local models available. "
@@ -258,76 +338,16 @@ def create_timmy(
         logger.info("Using fallback model %s (requested was unavailable)", model_name)
 
     use_tools = _model_supports_tools(model_name)
+    tools_list = _build_tools_list(use_tools, skip_mcp, model_name)
+    full_prompt = _build_prompt(use_tools, session_id)
 
-    # Conditionally include tools — small models get none
+    return _create_ollama_agent(
-    toolkit = create_full_toolkit() if use_tools else None
+        db_file=db_file,
-    if not use_tools:
+        model_name=model_name,
-        logger.info("Tools disabled for model %s (too small for reliable tool calling)", model_name)
+        tools_list=tools_list,
-
+        full_prompt=full_prompt,
-    # Build the tools list — Agno accepts a list of Toolkit / MCPTools
+        use_tools=use_tools,
-    tools_list: list = []
-    if toolkit:
-        tools_list.append(toolkit)
-
-    # Add MCP tool servers (lazy-connected on first arun()).
-    # Skipped when skip_mcp=True — MCP's stdio transport uses anyio cancel
-    # scopes that conflict with asyncio background task cancellation (#72).
-    if use_tools and not skip_mcp:
-        try:
-            from timmy.mcp_tools import create_filesystem_mcp_tools, create_gitea_mcp_tools
-
-            gitea_mcp = create_gitea_mcp_tools()
-            if gitea_mcp:
-                tools_list.append(gitea_mcp)
-
-            fs_mcp = create_filesystem_mcp_tools()
-            if fs_mcp:
-                tools_list.append(fs_mcp)
-        except Exception as exc:
-            logger.debug("MCP tools unavailable: %s", exc)
-
-    # Select prompt tier based on tool capability
-    base_prompt = get_system_prompt(tools_enabled=use_tools, session_id=session_id)
-
-    # Try to load memory context
-    try:
-        from timmy.memory_system import memory_system
-
-        memory_context = memory_system.get_system_context()
-        if memory_context:
-            # Truncate if too long — smaller budget for small models
-            # since the expanded prompt (roster, guardrails) uses more tokens
-            max_context = 2000 if not use_tools else 8000
-            if len(memory_context) > max_context:
-                memory_context = memory_context[:max_context] + "\n... [truncated]"
-            full_prompt = (
-                f"{base_prompt}\n\n"
-                f"## GROUNDED CONTEXT (verified sources — cite when using)\n\n"
-                f"{memory_context}"
-            )
-        else:
-            full_prompt = base_prompt
-    except Exception as exc:
-        logger.warning("Failed to load memory context: %s", exc)
-        full_prompt = base_prompt
-
-    model_kwargs = {}
-    if settings.ollama_num_ctx > 0:
-        model_kwargs["options"] = {"num_ctx": settings.ollama_num_ctx}
-    agent = Agent(
-        name="Agent",
-        model=Ollama(id=model_name, host=settings.ollama_url, timeout=300, **model_kwargs),
-        db=SqliteDb(db_file=db_file),
-        description=full_prompt,
-        add_history_to_context=True,
-        num_history_runs=20,
-        markdown=False,
-        tools=tools_list if tools_list else None,
-        tool_call_limit=settings.max_agent_steps if use_tools else None,
-        telemetry=settings.telemetry_enabled,
     )
-    _warmup_model(model_name)
-    return agent
 
 
 class TimmyWithMemory:

src/timmy/agentic_loop.py

@@ -95,6 +95,126 @@ def _parse_steps(plan_text: str) -> list[str]:
     return [line.strip() for line in plan_text.strip().splitlines() if line.strip()]
 
 
+# ---------------------------------------------------------------------------
+# Extracted helpers
+# ---------------------------------------------------------------------------
+
+
+def _extract_content(run_result) -> str:
+    """Extract text content from an agent run result."""
+    return run_result.content if hasattr(run_result, "content") else str(run_result)
+
+
+def _clean(text: str) -> str:
+    """Clean a model response using session's response cleaner."""
+    from timmy.session import _clean_response
+
+    return _clean_response(text)
+
+
+async def _plan_task(
+    agent, task: str, session_id: str, max_steps: int
+) -> tuple[list[str], bool] | str:
+    """Run the planning phase — returns (steps, was_truncated) or error string."""
+    plan_prompt = (
+        f"Break this task into numbered steps (max {max_steps}). "
+        f"Return ONLY a numbered list, nothing else.\n\n"
+        f"Task: {task}"
+    )
+    try:
+        plan_run = await asyncio.to_thread(
+            agent.run, plan_prompt, stream=False, session_id=f"{session_id}_plan"
+        )
+        plan_text = _extract_content(plan_run)
+    except Exception as exc:  # broad catch intentional: agent.run can raise any error
+        logger.error("Agentic loop: planning failed: %s", exc)
+        return f"Planning failed: {exc}"
+
+    steps = _parse_steps(plan_text)
+    if not steps:
+        return "Planning produced no steps."
+
+    planned_count = len(steps)
+    steps = steps[:max_steps]
+    return steps, planned_count > len(steps)
+
+
+async def _execute_step(
+    agent,
+    task: str,
+    step_desc: str,
+    step_num: int,
+    total_steps: int,
+    recent_results: list[str],
+    session_id: str,
+) -> AgenticStep:
+    """Execute a single step, returning an AgenticStep."""
+    step_start = time.monotonic()
+    context = (
+        f"Task: {task}\n"
+        f"Step {step_num}/{total_steps}: {step_desc}\n"
+        f"Recent progress: {recent_results[-2:] if recent_results else []}\n\n"
+        f"Execute this step and report what you did."
+    )
+    step_run = await asyncio.to_thread(
+        agent.run, context, stream=False, session_id=f"{session_id}_step{step_num}"
+    )
+    step_result = _clean(_extract_content(step_run))
+    return AgenticStep(
+        step_num=step_num,
+        description=step_desc,
+        result=step_result,
+        status="completed",
+        duration_ms=int((time.monotonic() - step_start) * 1000),
+    )
+
+
+async def _adapt_step(
+    agent,
+    step_desc: str,
+    step_num: int,
+    error: Exception,
+    step_start: float,
+    session_id: str,
+) -> AgenticStep:
+    """Attempt adaptation after a step failure."""
+    adapt_prompt = (
+        f"Step {step_num} failed with error: {error}\n"
+        f"Original step was: {step_desc}\n"
+        f"Adapt the plan and try an alternative approach for this step."
+    )
+    adapt_run = await asyncio.to_thread(
+        agent.run, adapt_prompt, stream=False, session_id=f"{session_id}_adapt{step_num}"
+    )
+    adapt_result = _clean(_extract_content(adapt_run))
+    return AgenticStep(
+        step_num=step_num,
+        description=f"[Adapted] {step_desc}",
+        result=adapt_result,
+        status="adapted",
+        duration_ms=int((time.monotonic() - step_start) * 1000),
+    )
+
+
+def _summarize(result: AgenticResult, total_steps: int, was_truncated: bool) -> None:
+    """Fill in summary and final status on the result object (mutates in place)."""
+    completed = sum(1 for s in result.steps if s.status == "completed")
+    adapted = sum(1 for s in result.steps if s.status == "adapted")
+    failed = sum(1 for s in result.steps if s.status == "failed")
+
+    parts = [f"Completed {completed}/{total_steps} steps"]
+    if adapted:
+        parts.append(f"{adapted} adapted")
+    if failed:
+        parts.append(f"{failed} failed")
+    result.summary = f"{result.task}: {', '.join(parts)}."
+
+    if was_truncated or len(result.steps) < total_steps or failed:
+        result.status = "partial"
+    else:
+        result.status = "completed"
+
+
 # ---------------------------------------------------------------------------
 # Core loop
 # ---------------------------------------------------------------------------
@@ -125,88 +245,41 @@ async def run_agentic_loop(
 
     task_id = str(uuid.uuid4())[:8]
     start_time = time.monotonic()
-
     agent = _get_loop_agent()
     result = AgenticResult(task_id=task_id, task=task, summary="")
 
-    # ── Phase 1: Planning ──────────────────────────────────────────────────
+    # Phase 1: Planning
-    plan_prompt = (
+    plan = await _plan_task(agent, task, session_id, max_steps)
-        f"Break this task into numbered steps (max {max_steps}). "
+    if isinstance(plan, str):
-        f"Return ONLY a numbered list, nothing else.\n\n"
-        f"Task: {task}"
-    )
-    try:
-        plan_run = await asyncio.to_thread(
-            agent.run, plan_prompt, stream=False, session_id=f"{session_id}_plan"
-        )
-        plan_text = plan_run.content if hasattr(plan_run, "content") else str(plan_run)
-    except Exception as exc:  # broad catch intentional: agent.run can raise any error
-        logger.error("Agentic loop: planning failed: %s", exc)
         result.status = "failed"
-        result.summary = f"Planning failed: {exc}"
+        result.summary = plan
         result.total_duration_ms = int((time.monotonic() - start_time) * 1000)
         return result
 
-    steps = _parse_steps(plan_text)
+    steps, was_truncated = plan
-    if not steps:
-        result.status = "failed"
-        result.summary = "Planning produced no steps."
-        result.total_duration_ms = int((time.monotonic() - start_time) * 1000)
-        return result
-
-    # Enforce max_steps — track if we truncated
-    planned_steps = len(steps)
-    steps = steps[:max_steps]
     total_steps = len(steps)
-    was_truncated = planned_steps > total_steps
 
-    # Broadcast plan
     await _broadcast_progress(
         "agentic.plan_ready",
-        {
+        {"task_id": task_id, "task": task, "steps": steps, "total": total_steps},
-            "task_id": task_id,
-            "task": task,
-            "steps": steps,
-            "total": total_steps,
-        },
     )
 
-    # ── Phase 2: Execution ─────────────────────────────────────────────────
+    # Phase 2: Execution
     completed_results: list[str] = []
-
     for i, step_desc in enumerate(steps, 1):
         step_start = time.monotonic()
-
-        recent = completed_results[-2:] if completed_results else []
-        context = (
-            f"Task: {task}\n"
-            f"Step {i}/{total_steps}: {step_desc}\n"
-            f"Recent progress: {recent}\n\n"
-            f"Execute this step and report what you did."
-        )
-
         try:
-            step_run = await asyncio.to_thread(
+            step = await _execute_step(
-                agent.run, context, stream=False, session_id=f"{session_id}_step{i}"
+                agent,
-            )
+                task,
-            step_result = step_run.content if hasattr(step_run, "content") else str(step_run)
+                step_desc,
-
+                i,
-            # Clean the response
+                total_steps,
-            from timmy.session import _clean_response
+                completed_results,
-
+                session_id,
-            step_result = _clean_response(step_result)
-
-            step = AgenticStep(
-                step_num=i,
-                description=step_desc,
-                result=step_result,
-                status="completed",
-                duration_ms=int((time.monotonic() - step_start) * 1000),
             )
             result.steps.append(step)
-            completed_results.append(f"Step {i}: {step_result[:200]}")
+            completed_results.append(f"Step {i}: {step.result[:200]}")
-
-            # Broadcast progress
             await _broadcast_progress(
                 "agentic.step_complete",
                 {
@@ -214,46 +287,18 @@ async def run_agentic_loop(
                     "step": i,
                     "total": total_steps,
                     "description": step_desc,
-                    "result": step_result[:200],
+                    "result": step.result[:200],
                 },
             )
-
             if on_progress:
                 await on_progress(step_desc, i, total_steps)
 
         except Exception as exc:  # broad catch intentional: agent.run can raise any error
             logger.warning("Agentic loop step %d failed: %s", i, exc)
-
-            # ── Adaptation: ask model to adapt ─────────────────────────────
-            adapt_prompt = (
-                f"Step {i} failed with error: {exc}\n"
-                f"Original step was: {step_desc}\n"
-                f"Adapt the plan and try an alternative approach for this step."
-            )
             try:
-                adapt_run = await asyncio.to_thread(
+                step = await _adapt_step(agent, step_desc, i, exc, step_start, session_id)
-                    agent.run,
-                    adapt_prompt,
-                    stream=False,
-                    session_id=f"{session_id}_adapt{i}",
-                )
-                adapt_result = (
-                    adapt_run.content if hasattr(adapt_run, "content") else str(adapt_run)
-                )
-                from timmy.session import _clean_response
-
-                adapt_result = _clean_response(adapt_result)
-
-                step = AgenticStep(
-                    step_num=i,
-                    description=f"[Adapted] {step_desc}",
-                    result=adapt_result,
-                    status="adapted",
-                    duration_ms=int((time.monotonic() - step_start) * 1000),
-                )
                 result.steps.append(step)
-                completed_results.append(f"Step {i} (adapted): {adapt_result[:200]}")
+                completed_results.append(f"Step {i} (adapted): {step.result[:200]}")
-
                 await _broadcast_progress(
                     "agentic.step_adapted",
                     {
@@ -262,46 +307,26 @@ async def run_agentic_loop(
                         "total": total_steps,
                         "description": step_desc,
                         "error": str(exc),
-                        "adaptation": adapt_result[:200],
+                        "adaptation": step.result[:200],
                     },
                 )
-
                 if on_progress:
                     await on_progress(f"[Adapted] {step_desc}", i, total_steps)
-
+            except Exception as adapt_exc:  # broad catch intentional
-            except Exception as adapt_exc:  # broad catch intentional: agent.run can raise any error
                 logger.error("Agentic loop adaptation also failed: %s", adapt_exc)
-                step = AgenticStep(
+                result.steps.append(
-                    step_num=i,
+                    AgenticStep(
-                    description=step_desc,
+                        step_num=i,
-                    result=f"Failed: {exc}; Adaptation also failed: {adapt_exc}",
+                        description=step_desc,
-                    status="failed",
+                        result=f"Failed: {exc}; Adaptation also failed: {adapt_exc}",
-                    duration_ms=int((time.monotonic() - step_start) * 1000),
+                        status="failed",
+                        duration_ms=int((time.monotonic() - step_start) * 1000),
+                    )
                 )
-                result.steps.append(step)
                 completed_results.append(f"Step {i}: FAILED")
 
-    # ── Phase 3: Summary ───────────────────────────────────────────────────
+    # Phase 3: Summary
-    completed_count = sum(1 for s in result.steps if s.status == "completed")
+    _summarize(result, total_steps, was_truncated)
-    adapted_count = sum(1 for s in result.steps if s.status == "adapted")
-    failed_count = sum(1 for s in result.steps if s.status == "failed")
-    parts = [f"Completed {completed_count}/{total_steps} steps"]
-    if adapted_count:
-        parts.append(f"{adapted_count} adapted")
-    if failed_count:
-        parts.append(f"{failed_count} failed")
-    result.summary = f"{task}: {', '.join(parts)}."
-
-    # Determine final status
-    if was_truncated:
-        result.status = "partial"
-    elif len(result.steps) < total_steps:
-        result.status = "partial"
-    elif any(s.status == "failed" for s in result.steps):
-        result.status = "partial"
-    else:
-        result.status = "completed"
-
     result.total_duration_ms = int((time.monotonic() - start_time) * 1000)
 
     await _broadcast_progress(

src/timmy/session.py

@@ -97,11 +97,6 @@ async def chat(message: str, session_id: str | None = None) -> str:
         The agent's response text.
     """
     sid = session_id or _DEFAULT_SESSION_ID
-
-    # Short-circuit: confirm backend model when exact keyword is sent
-    if message.strip() == "Qwe":
-        return "Confirmed: Qwe backend"
-
     agent = _get_agent()
     session_logger = get_session_logger()
 

src/timmy/thinking.py

@@ -232,6 +232,90 @@ class ThinkingEngine:
             return False  # Disabled — never idle
         return datetime.now(UTC) - self._last_input_time > timedelta(minutes=timeout)
 
+    def _build_thinking_context(self) -> tuple[str, str, list["Thought"]]:
+        """Assemble the context needed for a thinking cycle.
+
+        Returns:
+            (memory_context, system_context, recent_thoughts)
+        """
+        memory_context = self._load_memory_context()
+        system_context = self._gather_system_snapshot()
+        recent_thoughts = self.get_recent_thoughts(limit=5)
+        return memory_context, system_context, recent_thoughts
+
+    async def _generate_novel_thought(
+        self,
+        prompt: str | None,
+        memory_context: str,
+        system_context: str,
+        recent_thoughts: list["Thought"],
+    ) -> tuple[str | None, str]:
+        """Run the dedup-retry loop to produce a novel thought.
+
+        Returns:
+            (content, seed_type) — content is None if no novel thought produced.
+        """
+        seed_type: str = "freeform"
+
+        for attempt in range(self._MAX_DEDUP_RETRIES + 1):
+            if prompt:
+                seed_type = "prompted"
+                seed_context = f"Journal prompt: {prompt}"
+            else:
+                seed_type, seed_context = self._gather_seed()
+
+            continuity = self._build_continuity_context()
+
+            full_prompt = _THINKING_PROMPT.format(
+                memory_context=memory_context,
+                system_context=system_context,
+                seed_context=seed_context,
+                continuity_context=continuity,
+            )
+
+            try:
+                raw = await self._call_agent(full_prompt)
+            except Exception as exc:
+                logger.warning("Thinking cycle failed (Ollama likely down): %s", exc)
+                return None, seed_type
+
+            if not raw or not raw.strip():
+                logger.debug("Thinking cycle produced empty response, skipping")
+                return None, seed_type
+
+            content = raw.strip()
+
+            # Dedup: reject thoughts too similar to recent ones
+            if not self._is_too_similar(content, recent_thoughts):
+                return content, seed_type  # Good — novel thought
+
+            if attempt < self._MAX_DEDUP_RETRIES:
+                logger.info(
+                    "Thought too similar to recent (attempt %d/%d), retrying with new seed",
+                    attempt + 1,
+                    self._MAX_DEDUP_RETRIES + 1,
+                )
+            else:
+                logger.warning(
+                    "Thought still repetitive after %d retries, discarding",
+                    self._MAX_DEDUP_RETRIES + 1,
+                )
+                return None, seed_type
+
+        return None, seed_type
+
+    async def _process_thinking_result(self, thought: "Thought") -> None:
+        """Run all post-hooks after a thought is stored."""
+        self._maybe_check_memory()
+        await self._maybe_distill()
+        await self._maybe_file_issues()
+        await self._check_workspace()
+        self._maybe_check_memory_status()
+        self._update_memory(thought)
+        self._log_event(thought)
+        self._write_journal(thought)
+        await self._broadcast(thought)
+
     async def think_once(self, prompt: str | None = None) -> Thought | None:
         """Execute one thinking cycle.
 
@@ -257,91 +341,21 @@ class ThinkingEngine:
             )
             return None
 
-        memory_context = self._load_memory_context()
+        memory_context, system_context, recent_thoughts = self._build_thinking_context()
-        system_context = self._gather_system_snapshot()
-        recent_thoughts = self.get_recent_thoughts(limit=5)
-
-        content: str | None = None
-        seed_type: str = "freeform"
-
-        for attempt in range(self._MAX_DEDUP_RETRIES + 1):
-            if prompt:
-                seed_type = "prompted"
-                seed_context = f"Journal prompt: {prompt}"
-            else:
-                seed_type, seed_context = self._gather_seed()
-
-            continuity = self._build_continuity_context()
-
-            full_prompt = _THINKING_PROMPT.format(
-                memory_context=memory_context,
-                system_context=system_context,
-                seed_context=seed_context,
-                continuity_context=continuity,
-            )
-
-            try:
-                raw = await self._call_agent(full_prompt)
-            except Exception as exc:
-                logger.warning("Thinking cycle failed (Ollama likely down): %s", exc)
-                return None
-
-            if not raw or not raw.strip():
-                logger.debug("Thinking cycle produced empty response, skipping")
-                return None
-
-            content = raw.strip()
-
-            # Dedup: reject thoughts too similar to recent ones
-            if not self._is_too_similar(content, recent_thoughts):
-                break  # Good — novel thought
-
-            if attempt < self._MAX_DEDUP_RETRIES:
-                logger.info(
-                    "Thought too similar to recent (attempt %d/%d), retrying with new seed",
-                    attempt + 1,
-                    self._MAX_DEDUP_RETRIES + 1,
-                )
-                content = None  # Will retry
-            else:
-                logger.warning(
-                    "Thought still repetitive after %d retries, discarding",
-                    self._MAX_DEDUP_RETRIES + 1,
-                )
-                return None
 
+        content, seed_type = await self._generate_novel_thought(
+            prompt,
+            memory_context,
+            system_context,
+            recent_thoughts,
+        )
         if not content:
             return None
 
         thought = self._store_thought(content, seed_type)
         self._last_thought_id = thought.id
 
-        # Post-hook: check memory status periodically
+        await self._process_thinking_result(thought)
-        self._maybe_check_memory()
-
-        # Post-hook: distill facts from recent thoughts periodically
-        await self._maybe_distill()
-
-        # Post-hook: file Gitea issues for actionable observations
-        await self._maybe_file_issues()
-
-        # Post-hook: check workspace for new messages from Hermes
-        await self._check_workspace()
-
-        # Post-hook: proactive memory status audit
-        self._maybe_check_memory_status()
-
-        # Post-hook: update MEMORY.md with latest reflection
-        self._update_memory(thought)
-
-        # Log to swarm event system
-        self._log_event(thought)
-
-        # Append to daily journal file
-        self._write_journal(thought)
-
-        # Broadcast to WebSocket clients
-        await self._broadcast(thought)
 
         logger.info(
             "Thought [%s] (%s): %s",

tests/infrastructure/test_error_capture.py

@@ -5,9 +5,14 @@ from datetime import UTC, datetime, timedelta
 from unittest.mock import patch
 
 from infrastructure.error_capture import (
+    _create_bug_report,
     _dedup_cache,
+    _extract_origin,
     _get_git_context,
     _is_duplicate,
+    _log_error_event,
+    _record_to_session,
+    _send_error_notification,
     _stack_hash,
     capture_error,
 )
@@ -193,3 +198,87 @@ class TestCaptureError:
 
     def teardown_method(self):
         _dedup_cache.clear()
+
+
+class TestExtractOrigin:
+    """Test _extract_origin helper."""
+
+    def test_returns_file_and_line(self):
+        try:
+            _make_exception()
+        except ValueError as e:
+            filename, lineno = _extract_origin(e)
+        assert filename.endswith("test_error_capture.py")
+        assert lineno > 0
+
+    def test_no_traceback_returns_defaults(self):
+        exc = ValueError("no tb")
+        exc.__traceback__ = None
+        assert _extract_origin(exc) == ("unknown", 0)
+
+
+class TestLogErrorEvent:
+    """Test _log_error_event helper."""
+
+    def test_does_not_crash_when_event_log_missing(self):
+        try:
+            raise RuntimeError("log test")
+        except RuntimeError as e:
+            _log_error_event(e, "test", "abc123", "file.py", 42, {})
+
+
+class TestCreateBugReport:
+    """Test _create_bug_report helper."""
+
+    def test_returns_none_on_import_failure(self):
+        try:
+            raise RuntimeError("report test")
+        except RuntimeError as e:
+            with patch("infrastructure.error_capture.logger"):
+                result = _create_bug_report(e, "test", "abc", "f.py", 1, {}, "tb", None)
+        # Returns a task id or None depending on whether swarm is available
+        assert result is None or isinstance(result, str)
+
+
+class TestSendErrorNotification:
+    """Test _send_error_notification helper."""
+
+    def test_does_not_crash_on_notifier_failure(self):
+        try:
+            raise RuntimeError("notify test")
+        except RuntimeError as e:
+            _send_error_notification(e, "test")
+
+
+class TestRecordToSession:
+    """Test _record_to_session helper."""
+
+    def test_noop_when_no_recorder(self):
+        import infrastructure.error_capture as ec
+
+        original = ec._error_recorder
+        try:
+            ec._error_recorder = None
+            try:
+                raise RuntimeError("session test")
+            except RuntimeError as e:
+                _record_to_session(e, "test")  # should not crash
+        finally:
+            ec._error_recorder = original
+
+    def test_calls_registered_recorder(self):
+        import infrastructure.error_capture as ec
+
+        original = ec._error_recorder
+        calls = []
+        try:
+            ec._error_recorder = lambda **kwargs: calls.append(kwargs)
+            try:
+                raise RuntimeError("recorded")
+            except RuntimeError as e:
+                _record_to_session(e, "src")
+            assert len(calls) == 1
+            assert "RuntimeError: recorded" in calls[0]["error"]
+            assert calls[0]["context"] == "src"
+        finally:
+            ec._error_recorder = original

tests/timmy/test_agent.py

@@ -444,6 +444,150 @@ def test_get_effective_ollama_model_walks_fallback_chain():
         assert result == "fb-2"
 
 
+# ── _build_tools_list ─────────────────────────────────────────────────────
+
+
+def test_build_tools_list_empty_when_tools_disabled():
+    """Small models get an empty tools list."""
+    from timmy.agent import _build_tools_list
+
+    result = _build_tools_list(use_tools=False, skip_mcp=False, model_name="llama3.2")
+    assert result == []
+
+
+def test_build_tools_list_includes_toolkit_when_enabled():
+    """Tool-capable models get the full toolkit."""
+    mock_toolkit = MagicMock()
+    with patch("timmy.agent.create_full_toolkit", return_value=mock_toolkit):
+        from timmy.agent import _build_tools_list
+
+        result = _build_tools_list(use_tools=True, skip_mcp=True, model_name="llama3.1")
+        assert mock_toolkit in result
+
+
+def test_build_tools_list_skips_mcp_when_flagged():
+    """skip_mcp=True must not call MCP factories."""
+    mock_toolkit = MagicMock()
+    with (
+        patch("timmy.agent.create_full_toolkit", return_value=mock_toolkit),
+        patch("timmy.mcp_tools.create_gitea_mcp_tools") as mock_gitea,
+        patch("timmy.mcp_tools.create_filesystem_mcp_tools") as mock_fs,
+    ):
+        from timmy.agent import _build_tools_list
+
+        _build_tools_list(use_tools=True, skip_mcp=True, model_name="llama3.1")
+        mock_gitea.assert_not_called()
+        mock_fs.assert_not_called()
+
+
+def test_build_tools_list_includes_mcp_when_not_skipped():
+    """skip_mcp=False should attempt MCP tool creation."""
+    mock_toolkit = MagicMock()
+    with (
+        patch("timmy.agent.create_full_toolkit", return_value=mock_toolkit),
+        patch("timmy.mcp_tools.create_gitea_mcp_tools", return_value=None) as mock_gitea,
+        patch("timmy.mcp_tools.create_filesystem_mcp_tools", return_value=None) as mock_fs,
+    ):
+        from timmy.agent import _build_tools_list
+
+        _build_tools_list(use_tools=True, skip_mcp=False, model_name="llama3.1")
+        mock_gitea.assert_called_once()
+        mock_fs.assert_called_once()
+
+
+# ── _build_prompt ─────────────────────────────────────────────────────────
+
+
+def test_build_prompt_includes_base_prompt():
+    """Prompt should always contain the base system prompt."""
+    from timmy.agent import _build_prompt
+
+    result = _build_prompt(use_tools=False, session_id="test")
+    assert "Timmy" in result
+
+
+def test_build_prompt_appends_memory_context():
+    """Memory context should be appended when available."""
+    mock_memory = MagicMock()
+    mock_memory.get_system_context.return_value = "User prefers dark mode."
+    with patch("timmy.memory_system.memory_system", mock_memory):
+        from timmy.agent import _build_prompt
+
+        result = _build_prompt(use_tools=True, session_id="test")
+        assert "GROUNDED CONTEXT" in result
+        assert "dark mode" in result
+
+
+def test_build_prompt_truncates_long_memory():
+    """Long memory context should be truncated."""
+    mock_memory = MagicMock()
+    mock_memory.get_system_context.return_value = "x" * 10000
+    with patch("timmy.memory_system.memory_system", mock_memory):
+        from timmy.agent import _build_prompt
+
+        result = _build_prompt(use_tools=False, session_id="test")
+        assert "[truncated]" in result
+
+
+def test_build_prompt_survives_memory_failure():
+    """Prompt should fall back to base when memory fails."""
+    mock_memory = MagicMock()
+    mock_memory.get_system_context.side_effect = RuntimeError("db locked")
+    with patch("timmy.memory_system.memory_system", mock_memory):
+        from timmy.agent import _build_prompt
+
+        result = _build_prompt(use_tools=True, session_id="test")
+        assert "Timmy" in result
+        # Memory context should NOT be appended (the db locked error was caught)
+        assert "db locked" not in result
+
+
+# ── _create_ollama_agent ──────────────────────────────────────────────────
+
+
+def test_create_ollama_agent_passes_correct_kwargs():
+    """_create_ollama_agent must pass the expected kwargs to Agent."""
+    with (
+        patch("timmy.agent.Agent") as MockAgent,
+        patch("timmy.agent.Ollama"),
+        patch("timmy.agent.SqliteDb"),
+        patch("timmy.agent._warmup_model", return_value=True),
+    ):
+        from timmy.agent import _create_ollama_agent
+
+        _create_ollama_agent(
+            db_file="test.db",
+            model_name="llama3.1",
+            tools_list=[MagicMock()],
+            full_prompt="test prompt",
+            use_tools=True,
+        )
+        kwargs = MockAgent.call_args.kwargs
+        assert kwargs["description"] == "test prompt"
+        assert kwargs["markdown"] is False
+
+
+def test_create_ollama_agent_none_tools_when_empty():
+    """Empty tools_list should pass tools=None to Agent."""
+    with (
+        patch("timmy.agent.Agent") as MockAgent,
+        patch("timmy.agent.Ollama"),
+        patch("timmy.agent.SqliteDb"),
+        patch("timmy.agent._warmup_model", return_value=True),
+    ):
+        from timmy.agent import _create_ollama_agent
+
+        _create_ollama_agent(
+            db_file="test.db",
+            model_name="llama3.2",
+            tools_list=[],
+            full_prompt="test prompt",
+            use_tools=False,
+        )
+        kwargs = MockAgent.call_args.kwargs
+        assert kwargs["tools"] is None
+
+
 def test_no_hardcoded_fallback_constants_in_agent():
     """agent.py must not define module-level DEFAULT_MODEL_FALLBACKS."""
     import timmy.agent as agent_mod

tests/timmy/test_session.py

@@ -71,26 +71,6 @@ class TestAnnotateConfidence:
 # ---------------------------------------------------------------------------
 
 
-@pytest.mark.asyncio
-async def test_chat_confirms_qwe_backend():
-    """chat() should return exact confirmation when message is 'Qwe'."""
-    from timmy.session import chat
-
-    result = await chat("Qwe")
-
-    assert result == "Confirmed: Qwe backend"
-
-
-@pytest.mark.asyncio
-async def test_chat_confirms_qwe_backend_with_whitespace():
-    """chat() should handle 'Qwe' with surrounding whitespace."""
-    from timmy.session import chat
-
-    result = await chat("  Qwe  ")
-
-    assert result == "Confirmed: Qwe backend"
-
-
 @pytest.mark.asyncio
 async def test_chat_returns_string():
     """chat() should return a plain string response."""