zcbot/tests/test_skill_model_pinning.py

"""skill 定向模型(frontmatter `model:`)测试。

覆盖两层:
1. Skill.from_dir 解析 frontmatter 可选 `model` 字段(无 → "";有 → strip 后原样)。
2. AgentLoop._execute_tool_call 的热切挂点:load_skill 成功且 switcher 返回新
   (profile, caps, llm) → self.caps/self.llm 替换 + 结果尾追加说明 + emit model_switch;
   load_skill 报错 / switcher 抛异常 → 不切换、原模型继续。

switcher 本体(内置 skill 判定 + 档位豁免 + tasks 持久化)在 agent_builder 闭包里,
依赖 DB,不在此测;loop 侧只关心"给了切换指令就正确执行、失败不破坏 run"。
"""
from __future__ import annotations

import sys
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from core.loop import AgentLoop  # noqa: E402
from core.skills import Skill  # noqa: E402


def _write_skill(root: Path, name: str, frontmatter_extra: str = "") -> Path:
    d = root / name
    d.mkdir()
    (d / "SKILL.md").write_text(
        f"---\nname: {name}\ndescription: 测试用\n{frontmatter_extra}---\n\n# body\n",
        encoding="utf-8",
    )
    return d


class TestSkillModelField(unittest.TestCase):
    def setUp(self):
        self.tmpdir = tempfile.TemporaryDirectory()
        self.root = Path(self.tmpdir.name)

    def tearDown(self):
        self.tmpdir.cleanup()

    def test_no_model_field_defaults_empty(self):
        sk = Skill.from_dir(_write_skill(self.root, "plain"))
        self.assertEqual(sk.model, "")

    def test_model_field_parsed_and_stripped(self):
        sk = Skill.from_dir(_write_skill(self.root, "pinned", "model:  glm.pro52 \n"))
        self.assertEqual(sk.model, "glm.pro52")


class _FakeSession:
    task_id = "00000000-0000-0000-0000-000000000000"
    messages: list = []

    def append(self, msg):
        return 1


class _FakeExecutor:
    """load_skill 返回可配置文本;其余不需要。"""

    def __init__(self, result_text: str):
        self._result = result_text

    def schemas(self):
        return []

    def call_tool(self, name, args, ctx):
        return SimpleNamespace(content=self._result)


def _fake_caps(family="deepseek_v4", variant="flash"):
    return SimpleNamespace(
        family=family, variant=variant, max_iterations=10,
        default_reasoning_effort="", reliable_context=64000,
    )


def _make_loop(result_text: str, switcher) -> AgentLoop:
    loop = AgentLoop(
        llm=SimpleNamespace(tag="old_llm"),
        executor=_FakeExecutor(result_text),
        session=_FakeSession(),
        capabilities=_fake_caps(),
        user_id="00000000-0000-0000-0000-000000000000",
        working_dir=Path("."),
        skill_model_switch=switcher,
    )
    loop.events = []
    loop.sink = SimpleNamespace(emit=lambda ev: loop.events.append(ev))
    return loop


def _load_skill_tc(skill_name="ppt"):
    return SimpleNamespace(
        id="tc1",
        function=SimpleNamespace(name="load_skill", arguments=f'{{"name": "{skill_name}"}}'),
    )


class TestLoopHotSwap(unittest.TestCase):
    def test_switch_applies_new_caps_and_llm(self):
        new_caps = _fake_caps("glm", "pro52")
        new_llm = SimpleNamespace(tag="new_llm")
        calls = []

        def switcher(skill_name, cur_profile):
            calls.append((skill_name, cur_profile))
            return "glm.pro52", new_caps, new_llm

        loop = _make_loop("[skill=ppt, dir=x]\n# PPT", switcher)
        result, _ = loop._execute_tool_call(_load_skill_tc())

        self.assertEqual(calls, [("ppt", "deepseek_v4.flash")])
        self.assertIs(loop.caps, new_caps)
        self.assertIs(loop.llm, new_llm)
        self.assertIn("[模型切换]", result)
        self.assertIn("glm.pro52", result)
        types = [e["type"] for e in loop.events]
        self.assertIn("model_switch", types)
        ms = next(e for e in loop.events if e["type"] == "model_switch")
        self.assertEqual(ms["model_profile"], "glm.pro52")
        self.assertEqual(ms["from"], "deepseek_v4.flash")

    def test_no_switch_when_switcher_returns_none(self):
        loop = _make_loop("[skill=ppt, dir=x]\n# PPT", lambda n, c: None)
        old_caps, old_llm = loop.caps, loop.llm
        result, _ = loop._execute_tool_call(_load_skill_tc())
        self.assertIs(loop.caps, old_caps)
        self.assertIs(loop.llm, old_llm)
        self.assertNotIn("[模型切换]", result)
        self.assertNotIn("model_switch", [e["type"] for e in loop.events])

    def test_error_result_skips_switcher(self):
        calls = []
        loop = _make_loop(
            "[Error] skill 'nope' not found.",
            lambda n, c: calls.append(n),
        )
        loop._execute_tool_call(_load_skill_tc("nope"))
        self.assertEqual(calls, [])

    def test_switcher_exception_warns_and_keeps_model(self):
        def switcher(n, c):
            raise FileNotFoundError("glm.yaml missing")

        loop = _make_loop("[skill=ppt, dir=x]\n# PPT", switcher)
        old_caps, old_llm = loop.caps, loop.llm
        result, _ = loop._execute_tool_call(_load_skill_tc())
        self.assertIs(loop.caps, old_caps)
        self.assertIs(loop.llm, old_llm)
        self.assertNotIn("[模型切换]", result)
        warns = [e for e in loop.events if e["type"] == "warn"]
        self.assertTrue(any("切换失败" in e["msg"] for e in warns))

    def test_other_tools_never_trigger_switcher(self):
        calls = []
        loop = _make_loop("file content", lambda n, c: calls.append(n))
        tc = SimpleNamespace(
            id="tc2", function=SimpleNamespace(name="read", arguments='{"path": "a.txt"}')
        )
        loop._execute_tool_call(tc)
        self.assertEqual(calls, [])


if __name__ == "__main__":
    unittest.main()