refactor(tools): consolidate duplicated _truncate and get_mcp_manager into src/tool_utils (#3478)

* refactor(tools): consolidate duplicated _truncate and get_mcp_manager into src/tool_utils

Move all copies of _truncate(), get_mcp_manager(), and set_mcp_manager()
into a single leaf module (src/tool_utils.py) that imports only from
src.constants. This eliminates the lazy-import hack
('from src import agent_tools' inside function bodies) in tool_execution.py
and tool_implementations.py, and fixes a latent bug: the _truncate copy in
tool_execution.py was missing the isinstance guard and would crash on None.

Also deletes mcp_servers/_common.py — it was dead code with zero callers
anywhere in the codebase, containing its own copy of truncate() and
constants that already exist in src/constants.py.

* fix(tools): route remaining get_mcp_manager imports to src.tool_utils

The maintainer's feedback flagged src/task_scheduler.py:1857 and
routes/task_routes.py:977. A project-wide search found a third call site
in src/agent_loop.py that also imported get_mcp_manager from
src.agent_tools instead of src.tool_utils.

All three are now sourced from the canonical location in src.tool_utils.

---------

Co-authored-by: mcnoliveira <mcnoliveira@gmail.com>

mcp_servers/_common.py

@@ -1,22 +0,0 @@
-"""
-_common.py
-
-Shared constants and helpers for built-in MCP servers.
-"""
-
-MAX_OUTPUT_CHARS = 10_000
-MAX_READ_CHARS = 20_000
-SHELL_TIMEOUT = 60
-PYTHON_TIMEOUT = 30
-SEARCH_TIMEOUT = 30
-
-
-def truncate(text: str, limit: int = MAX_OUTPUT_CHARS) -> str:
-    """Truncate text to *limit* characters with a suffix note."""
-    if not isinstance(text, str):
-        # Tool output is occasionally None or a non-string; len(None) would
-        # raise. Coerce so this shared helper never crashes a tool response.
-        text = "" if text is None else str(text)
-    if len(text) > limit:
-        return text[:limit] + f"\n... (truncated, {len(text)} chars total)"
-    return text

routes/task_routes.py

@@ -974,7 +974,7 @@ def setup_task_routes(task_scheduler) -> APIRouter:
             "tag", "label", "move", "archive", "delete", "mark", "schedule",
         )
         try:
-            from src.agent_tools import get_mcp_manager
+            from src.tool_utils import get_mcp_manager
             mcp = get_mcp_manager()
             if mcp:
                 for tool in mcp.get_all_tools():

src/agent_loop.py

@@ -21,6 +21,7 @@ from src.settings import get_setting
 from src.prompt_security import untrusted_context_message
 from src.tool_security import blocked_tools_for_owner, plan_mode_disabled_tools
 from src.tool_policy import GUIDE_ONLY_DIRECTIVE, ToolPolicy
+from src.tool_utils import get_mcp_manager
 from src.agent_tools import (
     parse_tool_blocks,
     strip_tool_blocks,
@@ -29,7 +30,6 @@ from src.agent_tools import (
     set_active_document,
     set_active_model,
     function_call_to_tool_block,
-    get_mcp_manager,
     FUNCTION_TOOL_SCHEMAS,
     TOOL_TAGS,
     ToolBlock,

src/agent_tools.py

@@ -14,7 +14,7 @@ Sub-modules:
 import logging
 from collections import namedtuple
 
-from src.constants import MAX_OUTPUT_CHARS, MAX_READ_CHARS
+from src.tool_utils import _truncate, get_mcp_manager, set_mcp_manager
 
 logger = logging.getLogger(__name__)
 
@@ -64,33 +64,6 @@ TOOL_TAGS = {"bash", "python", "web_search", "web_fetch", "read_file", "write_fi
 
 ToolBlock = namedtuple("ToolBlock", ["tool_type", "content"])
 
-# ---------------------------------------------------------------------------
-# MCP Manager (kept here — used by execution and agent_loop)
-# ---------------------------------------------------------------------------
-_mcp_manager = None
-
-def set_mcp_manager(manager):
-    """Set the global MCP manager instance."""
-    global _mcp_manager
-    _mcp_manager = manager
-
-def get_mcp_manager():
-    """Get the global MCP manager instance."""
-    return _mcp_manager
-
-# ---------------------------------------------------------------------------
-# Helpers (kept here — used by sub-modules)
-# ---------------------------------------------------------------------------
-def _truncate(text: str, limit: int = MAX_OUTPUT_CHARS) -> str:
-    # Callers treat the result as text, so always return a string: coerce a
-    # non-string (None -> "", otherwise str(...)) instead of returning it raw,
-    # which would just move the crash downstream.
-    if not isinstance(text, str):
-        text = "" if text is None else str(text)
-    if len(text) > limit:
-        return text[:limit] + f"\n... (truncated, {len(text)} chars total)"
-    return text
-
 # ---------------------------------------------------------------------------
 # Re-exports from sub-modules
 # ---------------------------------------------------------------------------

src/task_scheduler.py

@@ -1098,7 +1098,7 @@ class TaskScheduler:
                                endpoint_url: str, model: str) -> str:
         """Gather raw data from all integrations, hand it to the LLM to write the check-in."""
         from src.tool_implementations import do_manage_notes
-        from src.agent_tools import get_mcp_manager
+        from src.tool_utils import get_mcp_manager
 
         tz_name = _resolve_task_timezone(db, task)
         try:
@@ -1854,7 +1854,7 @@ class TaskScheduler:
         have to special-case each tool's schema; the MCP tool ignores keys it
         doesn't recognise.
         """
-        from src.agent_tools import get_mcp_manager
+        from src.tool_utils import get_mcp_manager
         mcp = get_mcp_manager()
         if not mcp:
             logger.warning(f"Task {task.id}: MCP manager not available for delivery")

src/tool_execution.py

@@ -21,6 +21,7 @@ from typing import Any, Awaitable, Callable, Dict, Optional, Tuple
 from src.tool_security import is_public_blocked_tool, owner_is_admin_or_single_user
 from src.tool_policy import ToolPolicy
 from src.constants import MAX_OUTPUT_CHARS, MAX_READ_CHARS, MAX_DIFF_LINES, DATA_DIR
+from src.tool_utils import _truncate, get_mcp_manager
 
 # Persistent working directory for agent subprocesses.
 # Resolves to <repo_root>/data, which is the bind-mounted volume in Docker
@@ -326,12 +327,6 @@ PROGRESS_INTERVAL_S = 2.0
 # snippet without dragging the whole output along.
 PROGRESS_TAIL_LINES = 12
 
-
-def get_mcp_manager():
-    from src import agent_tools
-    return agent_tools.get_mcp_manager()
-
-
 # Directories ignored by the code-nav tools' Python fallbacks so results aren't
 # polluted by VCS internals / dependency trees / build caches. ripgrep already
 # honours .gitignore; this is the parity floor for the no-rg path (and the
@@ -364,12 +359,6 @@ def _resolve_search_root(raw_path: str, workspace: Optional[str] = None) -> str:
         return roots[0] if roots else os.path.realpath(".")
     return _resolve_tool_path(raw)
 
-
-def _truncate(text: str, limit: int = MAX_OUTPUT_CHARS) -> str:
-    if len(text) > limit:
-        return text[:limit] + f"\n... (truncated, {len(text)} chars total)"
-    return text
-
 logger = logging.getLogger(__name__)
 
 

src/tool_implementations.py

@@ -12,20 +12,10 @@ import os
 import re
 from typing import Any, Dict, List, Optional
 
-from src.constants import MAX_OUTPUT_CHARS, MAX_READ_CHARS, DEEP_RESEARCH_DIR, VAULT_FILE
+from src.constants import MAX_READ_CHARS, DEEP_RESEARCH_DIR, VAULT_FILE
+from src.tool_utils import get_mcp_manager
 from core.constants import internal_api_base
 
-
-def get_mcp_manager():
-    from src import agent_tools
-    return agent_tools.get_mcp_manager()
-
-
-def _truncate(text: str, limit: int = MAX_OUTPUT_CHARS) -> str:
-    if len(text) > limit:
-        return text[:limit] + f"\n... (truncated, {len(text)} chars total)"
-    return text
-
 logger = logging.getLogger(__name__)
 
 # ---------------------------------------------------------------------------

src/tool_utils.py

@@ -0,0 +1,39 @@
+"""
+This module intentionally imports NOTHING from the project (except
+src.constants which imports nothing from src). Adding a project import here
+will reintroduce the circular dependency that this module exists to break.
+"""
+
+from src.constants import MAX_OUTPUT_CHARS
+
+_mcp_manager = None
+
+# ---------------------------------------------------------------------------
+# MCP Manager singleton
+# ---------------------------------------------------------------------------
+
+def set_mcp_manager(manager):
+    """Set the global MCP manager instance."""
+    global _mcp_manager
+    _mcp_manager = manager
+
+def get_mcp_manager():
+    """Get the global MCP manager instance."""
+    return _mcp_manager
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+def _truncate(text: str, limit: int = MAX_OUTPUT_CHARS) -> str:
+    """
+    Truncate text to *limit* characters with a suffix note.
+
+    Callers treat the result as text, so always return a string: coerce a
+    non-string (None -> "", otherwise str(...)) instead of returning it raw,
+    which would just move the crash downstream.
+    """
+    if not isinstance(text, str):
+        text = "" if text is None else str(text)
+    if len(text) > limit:
+        return text[:limit] + f"\n... (truncated, {len(text)} chars total)"
+    return text

tests/test_mcp_common_truncate.py

@@ -1,27 +1,17 @@
-"""Regression: the shared MCP truncate() must tolerate non-string input."""
-import importlib.machinery
-import importlib.util
-from pathlib import Path
+"""Canonical _truncate must tolerate non-string input (regression).
 
-_PATH = Path(__file__).resolve().parents[1] / "mcp_servers" / "_common.py"
-
-
-def _load():
-    loader = importlib.machinery.SourceFileLoader("odysseus_mcp_common", str(_PATH))
-    spec = importlib.util.spec_from_loader(loader.name, loader)
-    module = importlib.util.module_from_spec(spec)
-    loader.exec_module(module)
-    return module
+Originally this tested mcp_servers/_common.py's copy, which was deleted
+since it had zero callers. Now it tests the canonical version.
+"""
 
+from src.tool_utils import _truncate
 
 def test_truncate_handles_none_and_nonstring():
-    c = _load()
-    assert c.truncate(None) == ""
-    assert c.truncate(12345) == "12345"
+    assert _truncate(None) == ""       # pyright: ignore[reportArgumentType]
+    assert _truncate(12345) == "12345" # pyright: ignore[reportArgumentType]
 
 
 def test_truncate_string_behaviour_unchanged():
-    c = _load()
-    assert c.truncate("hello", limit=100) == "hello"
-    out = c.truncate("x" * 50, limit=10)
+    assert _truncate("hello", limit=100) == "hello"
+    out = _truncate("x" * 50, limit=10)
     assert out.startswith("x" * 10) and "truncated" in out

tests/test_tool_utils_import_clean.py

@@ -0,0 +1,22 @@
+"""Verify src.tool_utils has no project imports beyond src.constants.
+
+If someone adds an import from src.settings, src.database, or any other
+project module inside tool_utils.py, the circular import that this module
+exists to break will silently return a partially-initialized module.
+This test catches that statically.
+"""
+
+import ast
+import pathlib
+
+
+def test_tool_utils_has_no_project_imports():
+    src = pathlib.Path("src/tool_utils.py").read_text()
+    tree = ast.parse(src)
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.Import, ast.ImportFrom)):
+            if isinstance(node, ast.ImportFrom) and node.module:
+                msg = f"Illegal project import in tool_utils.py: {node.module}"
+                assert node.module in ("src.constants",) or not node.module.startswith(
+                    "src."
+                ), msg