Harden MCP config loading and action-state retries

src/octopal/infrastructure/mcp/manager.py

@@ -50,21 +50,32 @@ class MCPServerConfig:
     "upstream_5xx",
     "unknown_error",
 }
-_MCP_SLOW_TOOL_HINTS = (
-    "search",
-    "fetch",
-    "crawl",
+_MCP_SLOW_TOOL_HINTS = (
+    "search",
+    "fetch",
+    "crawl",
     "thread",
     "inbox",
     "mail",
     "list_",
-    "query",
-)
-
-class MCPManager:
+    "query",
+)
+
+
+def _extract_mcp_server_configs(config_data: Any) -> Any:
+    if not isinstance(config_data, dict):
+        return None
+    if isinstance(config_data.get("servers"), dict):
+        return config_data["servers"]
+    if isinstance(config_data.get("mcpServers"), dict):
+        return config_data["mcpServers"]
+    return config_data
+
+
+class MCPManager:
     def __init__(self, workspace_dir: Path):
         self.workspace_dir = workspace_dir
-        self.sessions: dict[str, ClientSession] = {}
+        self.sessions: dict[str, ClientSession] = {}
         # Stores the background task that keeps the session alive
         self._tasks: dict[str, asyncio.Task] = {}
         # Communication queues for disconnect signals
@@ -79,12 +90,19 @@ def __init__(self, workspace_dir: Path):
         self._shutdown_requested = False
         self.config_path = workspace_dir / "mcp_servers.json"
         self.legacy_config_path = workspace_dir / "config" / "mcp.json"
+        self.root_config_path = workspace_dir / "mcp.json"
+        self.claude_config_path = workspace_dir / ".mcp.json"
         self._configs_loaded = False
 
     def _config_paths(self) -> list[Path]:
-        # Keep supporting the newer flat mcp_servers.json layout while also
-        # reading the legacy workspace/config/mcp.json workspace file.
-        return [self.legacy_config_path, self.config_path]
+        # Read broad compatibility paths first, then let canonical Octopal
+        # config files override duplicates from imported MCP client configs.
+        return [
+            self.claude_config_path,
+            self.root_config_path,
+            self.legacy_config_path,
+            self.config_path,
+        ]
 
     def _load_configs_from_disk(self) -> dict[str, MCPServerConfig]:
         if self._configs_loaded:
@@ -97,8 +115,16 @@ def _load_configs_from_disk(self) -> dict[str, MCPServerConfig]:
         for config_path in self._config_paths():
             if not config_path.exists():
                 continue
-            config_data = json.loads(config_path.read_text(encoding="utf-8"))
-            server_configs = config_data.get("servers", config_data)
+            config_text = config_path.read_text(encoding="utf-8").strip()
+            if not config_text:
+                logger.warning("Skipping empty MCP config file", path=str(config_path))
+                continue
+            try:
+                config_data = json.loads(config_text)
+            except json.JSONDecodeError:
+                logger.warning("Skipping invalid MCP config file", path=str(config_path), exc_info=True)
+                continue
+            server_configs = _extract_mcp_server_configs(config_data)
             if not isinstance(server_configs, dict):
                 continue
             for server_id, cfg in server_configs.items():

src/octopal/runtime/octo/router.py

@@ -1546,9 +1546,6 @@ def _looks_like_internal_worker_followup_leak(text: str) -> bool:
         "current tool set",
         "bounded route",
         "on the next turn",
-        "полный режим",
-        "нет доступа к a2a",
-        "нет `a2a_send_message`",
     )
     orchestration_phrases = (
         "can't modify",
@@ -3647,6 +3644,8 @@ async def _needs_action_or_blocked_retry(
 ) -> bool:
     if not normalize_plain_text(candidate or "") or should_suppress_user_delivery(candidate):
         return False
+    if _looks_like_unbacked_action_commitment(candidate):
+        return True
     prompt = (
         "Classify whether the draft assistant response is safe to deliver as the final answer for this turn.\n"
         "Return JSON only with this shape:\n"
@@ -3684,6 +3683,37 @@ async def _needs_action_or_blocked_retry(
     return verdict == "requires_runtime_action_state" and confidence >= 0.55
 
 
+_UNBACKED_ACTION_BLOCKED_MARKERS = (
+    "blocked",
+    "cannot",
+    "can't",
+    "could not",
+    "i need",
+    "need you",
+    "please confirm",
+)
+
+_UNBACKED_ACTION_PATTERNS = (
+    re.compile(
+        r"\b(?:i(?:'m| am)\s+)?("
+        r"installing|activating|creating|starting|running|connecting|configuring|"
+        r"restarting|updating|saving|adding|writing|checking"
+        r")\b",
+        re.IGNORECASE,
+    ),
+)
+
+
+def _looks_like_unbacked_action_commitment(candidate: str) -> bool:
+    cleaned = normalize_plain_text(candidate or "")
+    if not cleaned:
+        return False
+    lowered = cleaned.lower()
+    if any(marker in lowered for marker in _UNBACKED_ACTION_BLOCKED_MARKERS):
+        return False
+    return any(pattern.search(cleaned) for pattern in _UNBACKED_ACTION_PATTERNS)
+
+
 async def _verify_final_response(
     provider: InferenceProvider,
     messages: list[Message | dict[str, Any]],

src/octopal/tools/catalog.py

@@ -1334,10 +1334,10 @@ def get_tools(mcp_manager=None) -> list[ToolSpec]:
             parameters={
                 "type": "object",
                 "properties": {
-                    "path": {
-                        "type": "string",
-                        "description": "Workspace-relative path to write.",
-                    },
+                    "path": {
+                        "type": "string",
+                        "description": "Workspace-relative path to write, for example config/mcp.json instead of workspace/config/mcp.json.",
+                    },
                     "content": {"type": "string", "description": "File contents."},
                 },
                 "required": ["path", "content"],

src/octopal/tools/filesystem/path_safety.py

@@ -23,6 +23,7 @@ def resolve_workspace_path(
         raise WorkspacePathError("path contains null byte")
 
     root = base_dir.resolve()
+    raw = _strip_redundant_workspace_prefix(raw, root)
     candidate = Path(os.path.normpath(str(root / raw)))
     _assert_within(root, candidate)
 
@@ -37,19 +38,19 @@ def resolve_workspace_path(
         for p in allowed_paths:
             p_path = (root / p).resolve(strict=False)
             allowed_resolved.append(p_path)
-            
+
         # Check if candidate is within any of the allowed paths
         # or if it's the worker's own specific directory which isn't part of allowed_paths but should be implicit
         # Actually, worker's directory check is handled implicitly if it's not restricted or we just check allowed_paths
-        
+
         # We need to resolve candidate strictly if it exists, or loosely if it doesn't
         resolved_candidate = candidate.resolve(strict=False)
         is_allowed = False
         for allowed_root in allowed_resolved:
             if resolved_candidate == allowed_root or allowed_root in resolved_candidate.parents:
                 is_allowed = True
                 break
-                
+
         if not is_allowed:
             raise WorkspacePathError(f"access denied: path '{raw}' is outside allowed paths")
 
@@ -96,3 +97,15 @@ def _assert_existing_ancestor_within(root: Path, candidate: Path) -> None:
     if ancestor.exists():
         resolved_ancestor = ancestor.resolve(strict=True)
         _assert_within(root, resolved_ancestor)
+
+
+def _strip_redundant_workspace_prefix(raw: str, root: Path) -> str:
+    if Path(raw).is_absolute():
+        return raw
+    normalized = raw.replace("\\", "/")
+    if normalized.startswith("/"):
+        return raw
+    parts = [part for part in normalized.split("/") if part and part != "."]
+    if root.name == "workspace" and parts[:1] == ["workspace"]:
+        return "/".join(parts[1:]) or "."
+    return raw

tests/test_filesystem_hardening.py

@@ -51,6 +51,19 @@ def test_fs_write_rejects_symlink_escape(tmp_path: Path) -> None:
     assert not (outside / "pwn.txt").exists()
 
 
+def test_fs_write_treats_leading_workspace_as_redundant_at_workspace_root(
+    tmp_path: Path,
+) -> None:
+    workspace = tmp_path / "workspace"
+    workspace.mkdir(parents=True, exist_ok=True)
+
+    result = fs_write({"path": "workspace/config/mcp.json", "content": "{}"}, workspace)
+
+    assert result == "fs_write ok"
+    assert (workspace / "config" / "mcp.json").read_text(encoding="utf-8") == "{}"
+    assert not (workspace / "workspace" / "config" / "mcp.json").exists()
+
+
 def test_fs_delete_unlinks_symlink_without_touching_target(tmp_path: Path) -> None:
     _ensure_symlink_supported(tmp_path)
 

tests/test_mcp_manager.py

@@ -235,6 +235,55 @@ def test_mcp_manager_reads_legacy_workspace_mcp_config(tmp_path) -> None:
     assert resolved == ["AgentMail"]
 
 
+def test_mcp_manager_reads_claude_style_mcp_servers_config(tmp_path) -> None:
+    (tmp_path / ".mcp.json").write_text(
+        """
+{
+  "mcpServers": {
+    "minimax": {
+      "command": "uvx",
+      "args": ["minimax-coding-plan-mcp"],
+      "tools": ["mcp_minimax_web_search"]
+    }
+  }
+}
+""".strip(),
+        encoding="utf-8",
+    )
+
+    manager = MCPManager(tmp_path)
+
+    resolved = manager.resolve_configured_server_ids_for_tools(["mcp_minimax_web_search"])
+
+    assert resolved == ["minimax"]
+
+
+def test_mcp_manager_skips_empty_compat_config_and_reads_canonical(tmp_path) -> None:
+    (tmp_path / ".mcp.json").write_text("", encoding="utf-8")
+    config_dir = tmp_path / "config"
+    config_dir.mkdir()
+    (config_dir / "mcp.json").write_text(
+        """
+{
+  "servers": {
+    "docs": {
+      "command": "uvx",
+      "args": ["docs-mcp"],
+      "tools": ["mcp_docs_search"]
+    }
+  }
+}
+""".strip(),
+        encoding="utf-8",
+    )
+
+    manager = MCPManager(tmp_path)
+
+    resolved = manager.resolve_configured_server_ids_for_tools(["mcp_docs_search"])
+
+    assert resolved == ["docs"]
+
+
 def test_resolve_configured_server_id_for_tool_name_is_case_insensitive(tmp_path) -> None:
     configs = {
         "AgentMail": MCPServerConfig(