Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 31 additions & 12 deletions sdk/nexent/core/tools/read_skill_md_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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)
Expand Down
66 changes: 66 additions & 0 deletions test/sdk/core/tools/test_read_skill_md_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -273,6 +273,33 @@
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"):

Check warning on line 300 in test/sdk/core/tools/test_read_skill_md_tool.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this exception test to have only one invocation possibly throwing an exception.

See more on https://sonarcloud.io/project/issues?id=ModelEngine-Group_nexent&issues=AaA9OfK8LdDhLAEZWrps&open=AaA9OfK8LdDhLAEZWrps&pullRequest=3784
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()
Expand Down Expand Up @@ -361,6 +388,35 @@
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
Expand Down Expand Up @@ -446,6 +502,16 @@
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)
Expand Down
Loading