diff --git a/sdk/nexent/core/tools/read_skill_md_tool.py b/sdk/nexent/core/tools/read_skill_md_tool.py index 92f652e08..97f110cee 100644 --- a/sdk/nexent/core/tools/read_skill_md_tool.py +++ b/sdk/nexent/core/tools/read_skill_md_tool.py @@ -8,6 +8,31 @@ logger = logging.getLogger(__name__) +_SUPPORTED_ENCODINGS = ("utf-8-sig", "gb18030") + + +def _read_text_file(file_path: str) -> str: + """Read text using strict, deterministic encoding fallbacks. + + UTF-8 is preferred for skill files. GB18030 is the fallback because it is a + superset of GBK and GB2312. Decoding remains strict so unsupported or + corrupted input is reported instead of being replaced silently. + """ + with open(file_path, "rb") as file: + data = file.read() + + decode_errors = [] + for encoding in _SUPPORTED_ENCODINGS: + try: + return data.decode(encoding, errors="strict") + except UnicodeDecodeError as exc: + decode_errors.append(exc) + + attempted_encodings = ", ".join(_SUPPORTED_ENCODINGS) + raise UnicodeError( + f"Unable to decode '{file_path}' using supported encodings: {attempted_encodings}" + ) from decode_errors[-1] + class ReadSkillMdTool(Tool): """Tool for reading skill markdown files.""" @@ -86,16 +111,11 @@ def _read_skill_file(self, skill_dir: str, file_path: str) -> Tuple[str, bool]: for path in possible_paths: full_path = os.path.join(skill_dir, path) if os.path.exists(full_path): - try: - with open(full_path, 'r', encoding='utf-8') as f: - content = f.read() - # Strip frontmatter if it's a markdown file - if full_path.endswith('.md'): - content = self._strip_frontmatter(content) - return content, True - except Exception as e: - logger.warning(f"Failed to read file {path}: {e}") - continue + content = _read_text_file(full_path) + # Strip frontmatter if it's a markdown file + if full_path.endswith('.md'): + content = self._strip_frontmatter(content) + return content, True return f"File not found: {file_path}", False @@ -188,8 +208,7 @@ def _read_direct_file(self, path_parts: tuple) -> str: return f"File not found: {file_path}" try: - with open(full_path, 'r', encoding='utf-8') as f: - content = f.read() + content = _read_text_file(full_path) # Strip frontmatter if it's a markdown file if full_path.endswith('.md'): content = self._strip_frontmatter(content) diff --git a/test/sdk/core/tools/test_read_skill_md_tool.py b/test/sdk/core/tools/test_read_skill_md_tool.py index e9dc0eba8..d9a96b091 100644 --- a/test/sdk/core/tools/test_read_skill_md_tool.py +++ b/test/sdk/core/tools/test_read_skill_md_tool.py @@ -273,6 +273,33 @@ def test_read_existing_file(self, sample_skill): assert found is True assert "Skill Content" in content + @pytest.mark.parametrize("encoding", ["utf-8", "gbk", "gb18030"]) + def test_read_markdown_with_supported_encoding(self, temp_skills_dir, encoding): + """Test reading UTF-8 and common Chinese legacy encodings.""" + skill_dir = os.path.join(temp_skills_dir, f"{encoding}-skill") + os.makedirs(skill_dir) + expected_content = "# 技能说明\n这是编码读取测试，扩展字符：𠀀。" + if encoding == "gbk": + expected_content = "# 技能说明\n这是 GBK 编码读取测试。" + + with open(os.path.join(skill_dir, "SKILL.md"), "wb") as file: + file.write(expected_content.encode(encoding)) + + content, found = ReadSkillMdTool()._read_skill_file(skill_dir, "SKILL.md") + + assert found is True + assert content == expected_content + + def test_read_file_reports_unsupported_encoding(self, temp_skills_dir): + """Test that genuine decoding failures are not treated as missing files.""" + skill_dir = os.path.join(temp_skills_dir, "invalid-encoding-skill") + os.makedirs(skill_dir) + with open(os.path.join(skill_dir, "SKILL.md"), "wb") as file: + file.write(b"\xff\xff\xff") + + with pytest.raises(UnicodeError, match="utf-8-sig, gb18030"): + ReadSkillMdTool()._read_skill_file(skill_dir, "SKILL.md") + def test_read_file_with_extension(self, sample_skill): """Test reading a file with .md extension when not provided.""" tool = ReadSkillMdTool() @@ -361,6 +388,35 @@ def test_execute_reads_default_skill_md(self, sample_skill, temp_skills_dir): result = tool.execute(skill_name) assert "test-skill" in result.lower() or "Skill Content" in result + def test_execute_reads_gbk_skill_md(self, temp_skills_dir): + """Test execute reads a GBK-encoded default SKILL.md.""" + skill_name = "gbk-skill" + skill_dir = os.path.join(temp_skills_dir, skill_name) + os.makedirs(skill_dir) + expected_content = "# 技能说明\n这是 GBK 编码的技能文件。" + with open(os.path.join(skill_dir, "SKILL.md"), "wb") as file: + file.write(expected_content.encode("gbk")) + + tool = ReadSkillMdTool(local_skills_dir=temp_skills_dir) + tool.skill_manager = MockSkillManager(temp_skills_dir) + + assert tool.execute(skill_name) == expected_content + + def test_execute_reports_decode_error(self, temp_skills_dir): + """Test execute distinguishes decoding errors from missing files.""" + skill_name = "invalid-encoding-skill" + skill_dir = os.path.join(temp_skills_dir, skill_name) + os.makedirs(skill_dir) + with open(os.path.join(skill_dir, "SKILL.md"), "wb") as file: + file.write(b"\xff\xff\xff") + + tool = ReadSkillMdTool(local_skills_dir=temp_skills_dir) + tool.skill_manager = MockSkillManager(temp_skills_dir) + result = tool.execute(skill_name) + + assert "Unable to decode" in result + assert "not found" not in result.lower() + def test_execute_reads_additional_files(self, sample_skill_with_files, temp_skills_dir): """Test execute reads specified additional files.""" skill_dir, skill_name = sample_skill_with_files @@ -446,6 +502,16 @@ def test_read_direct_file_with_path(self, temp_skills_dir): result = tool._read_direct_file(("test-file.txt",)) assert "test content" in result + def test_read_direct_gbk_file(self, temp_skills_dir): + """Test direct reads use the same GBK-compatible fallback.""" + tool = ReadSkillMdTool(local_skills_dir=temp_skills_dir) + tool.skill_manager = MockSkillManager(temp_skills_dir) + expected_content = "根目录下的 GBK 文件" + with open(os.path.join(temp_skills_dir, "legacy.txt"), "wb") as file: + file.write(expected_content.encode("gbk")) + + assert tool._read_direct_file(("legacy.txt",)) == expected_content + def test_read_direct_file_not_found(self, temp_skills_dir): """Test _read_direct_file returns error for missing file.""" tool = ReadSkillMdTool(local_skills_dir=temp_skills_dir)