"""Transport-only skill runtime tests: no bot imports, filesystem, or network.""" import asyncio import json from pathlib import Path import sys import unittest from unittest.mock import AsyncMock, Mock, patch sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "bot-utility")) import skill_runtime as runtime class Snapshot: def __init__(self, *, enabled=True, records=None, always=(), scenes=None, limits=None): self.enabled = enabled self.records = records if records is not None else { "alpha": {"name": "alpha", "description": "Alpha guidance", "body": "EXACT alpha body\n第二行", "references": {"references/note.md": "EXACT reference\n不要截断"}}, "beta": {"name": "beta", "description": "Beta guidance", "body": "EXACT beta body", "references": {}}, "gamma": {"name": "gamma", "description": "Gamma guidance", "body": "EXACT gamma body", "references": {}}, "delta": {"name": "delta", "description": "Delta guidance", "body": "EXACT delta body", "references": {}}, } self.always = always self.scenes = scenes or {} self.limits = {"max_rounds": 2, "max_skills": 3, "max_tokens": 12000, **(limits or {})} self.fingerprint = "fake-pinned-fingerprint" self.loads = [] self.reads = [] def catalog_text(self): return "\n".join(name + ": " + self.records[name]["description"] for name in sorted(self.records)) def bound_names(self, scene): bindings = self.scenes.get(scene, ()) if scene == "reminder": bindings = self.scenes.get("notification", ()) + bindings return tuple(dict.fromkeys(self.always + bindings)) def load(self, name): self.loads.append(name) return self.records[name]["body"] def read(self, name, path): self.reads.append((name, path)) return self.records[name]["references"][path] def tool(action="load", **params): return {"name": "skill", "args": {"action": action, "params": params}} def chat(*tools, content="final"): return json.dumps({"message": [{"content": content}], "tools": list(tools)}, ensure_ascii=False) def text(*tools, content="final text"): return json.dumps({"text": content, "tools": list(tools)}, ensure_ascii=False) class SkillRuntimeTests(unittest.IsolatedAsyncioTestCase): def setUp(self): self.snapshot = Snapshot() self.registry = Mock() self.registry.snapshot.return_value = self.snapshot self.patch = patch.object(runtime, "registry", self.registry) self.patch.start() self.addCleanup(self.patch.stop) print_patch = patch.object(runtime, "print", create=True) self.summary_print = print_patch.start() self.addCleanup(print_patch.stop) async def test_final_prose_reuses_pinned_styles_without_json_or_tools(self): self.snapshot.always = ('alpha',) self.snapshot.scenes = {'tool_result': ('beta',)} with runtime.skill_session('private_chat') as turn: material = runtime.loaded_style_context('tool_result') self.assertIn('EXACT alpha body', material) self.assertIn('EXACT beta body', material) self.assertNotIn('[Skill output contract]', material) self.assertNotIn('[Full skill catalog]', material) self.assertEqual(turn.rounds, 0) call = AsyncMock(return_value='自然回复') with runtime.suspend_skills(): result = await runtime.generate_with_skills(call, material, 'user', reasoning=False) self.assertEqual(result, '自然回复') self.assertIs(runtime.current_turn(), turn) self.registry.snapshot.assert_called_once() self.assertEqual(runtime.loaded_style_context(), '') async def test_no_context_exact_identity_passthrough(self): system, user, images, answer = object(), object(), object(), object() call = AsyncMock(return_value=answer) result = await runtime.generate_with_skills(call, system, user, 7, presence_penalty=.7, image_infos=images, reasoning=False) self.assertIs(result, answer) call.assert_awaited_once_with(system, user, 7, presence_penalty=.7, image_infos=images, reasoning=False) self.registry.snapshot.assert_not_called() async def test_disabled_and_empty_exact_passthrough(self): for snapshot in (Snapshot(enabled=False), Snapshot(records={})): self.registry.snapshot.return_value = snapshot call = AsyncMock(return_value="not JSON and unmodified") with runtime.skill_session("help", "text"): self.assertEqual(runtime.estimated_overhead(), 0) self.assertEqual(await runtime.generate_with_skills(call, "system", "user"), "not JSON and unmodified") call.assert_awaited_once_with("system", "user", 0) async def test_mixed_plan_discarded_and_transport_kwargs_preserved(self): business = {"name": "send_mail", "args": {"body": "PROVISIONAL SECRET"}} final = chat({"name": "web_search", "args": {"query": "safe"}}) call = AsyncMock(side_effect=[chat(tool(name="alpha"), business, content="PROVISIONAL"), final]) images = [{"url": "image"}] with runtime.skill_session("private_chat"): result = await runtime.generate_with_skills(call, "ORIGINAL SYSTEM", "ORIGINAL USER", 4, image_infos=images, presence_penalty=.5, reasoning=False) self.assertEqual(result, final) self.assertEqual(call.await_count, 2) second = call.await_args_list[1] self.assertIn(self.snapshot.records["alpha"]["body"], second.args[0]) self.assertNotIn("PROVISIONAL", second.args[0]) self.assertIn("ORIGINAL SYSTEM", second.args[0]) self.assertEqual(second.args[1:], ("ORIGINAL USER", 4)) self.assertIs(second.kwargs["image_infos"], images) self.assertEqual(second.kwargs["presence_penalty"], .5) self.assertTrue(second.kwargs["reasoning"]) async def test_load_then_read_exact_reference_and_final_disabled(self): call = AsyncMock(side_effect=[chat(tool(name="alpha")), chat(tool("read", name="alpha", path="references/note.md")), chat()]) with runtime.skill_session("private_chat") as turn: self.assertEqual(await runtime.generate_with_skills(call, "S", "U"), chat()) self.assertEqual(turn.rounds, 2) prompt = call.await_args.args[0] self.assertIn(self.snapshot.records["alpha"]["body"], prompt) self.assertIn(self.snapshot.records["alpha"]["references"]["references/note.md"], prompt) self.assertIn("skill tools disabled", prompt) self.assertEqual(self.snapshot.reads, [("alpha", "references/note.md")]) async def test_read_requires_loaded_skill_and_invalid_path_safe(self): call = AsyncMock(side_effect=[chat(tool("read", name="alpha", path="references/note.md")), chat(tool(name="alpha"), tool("read", name="alpha", path="../../secrets")), chat()]) with runtime.skill_session("private_chat"): await runtime.generate_with_skills(call, "S", "U") prompt = call.await_args.args[0] self.assertIn("skill_not_loaded", prompt) self.assertIn("invalid_reference", prompt) self.assertNotIn("../../secrets", prompt) self.assertEqual(len(self.snapshot.reads), 1) async def test_duplicate_material_not_injected_twice(self): self.snapshot.always = ("alpha",) call = AsyncMock(side_effect=[chat(tool(name="alpha"), tool("read", name="alpha", path="references/note.md"), tool("read", name="alpha", path="references/note.md")), chat()]) with runtime.skill_session("private_chat"): await runtime.generate_with_skills(call, "S", "U") await runtime.generate_with_skills(AsyncMock(return_value=chat()), "S", "U", reasoning=True) self.assertEqual(self.snapshot.loads, ["alpha"]) self.assertEqual(len(self.snapshot.reads), 1) prompt = call.await_args.args[0] self.assertEqual(prompt.count("EXACT alpha body"), 1) self.assertEqual(prompt.count("EXACT reference"), 1) async def test_repeat_read_loop_fails_closed_after_one_final_call(self): call = AsyncMock(return_value=chat(tool(name="alpha"), {"name": "send_mail"})) with runtime.skill_session("private_chat") as turn: self.assertIsNone(await runtime.generate_with_skills(call, "S", "U")) self.assertTrue(turn.failed) self.assertIsNone(await runtime.generate_with_skills(call, "S", "U")) self.assertEqual(call.await_count, 3) self.assertEqual(self.snapshot.loads, ["alpha"]) self.assertTrue(all(item.kwargs["reasoning"] for item in call.await_args_list[1:])) async def test_skill_count_includes_scene_and_always_bindings(self): self.snapshot.always = ("alpha",) self.snapshot.scenes = {"private_chat": ("beta", "gamma")} call = AsyncMock(side_effect=[chat(tool(name="delta")), chat()]) with runtime.skill_session("private_chat") as turn: await runtime.generate_with_skills(call, "S", "U") self.assertEqual(set(turn.loaded), {"alpha", "beta", "gamma"}) self.assertIn("skill_budget", call.await_args.args[0]) self.assertNotIn("EXACT delta body", call.await_args.args[0]) async def test_budget_rejects_whole_body_not_truncated(self): huge = "大" * 12500 self.snapshot.records["alpha"]["body"] = huge call = AsyncMock(side_effect=[chat(tool(name="alpha")), chat()]) with runtime.skill_session("private_chat") as turn: await runtime.generate_with_skills(call, "S", "U") self.assertNotIn("alpha", turn.loaded) prompt = call.await_args.args[0] self.assertIn('"reason": "token_budget"', prompt) self.assertNotIn("大", prompt) self.assertGreater(runtime.estimate_tokens("中" * 10), runtime.estimate_tokens("a" * 10)) async def test_reference_budget_rejects_whole_reference(self): self.snapshot.always = ("alpha",) self.snapshot.records["alpha"]["references"]["references/huge.md"] = "巨" * 12000 call = AsyncMock(side_effect=[chat(tool("read", name="alpha", path="references/huge.md")), chat()]) with runtime.skill_session("private_chat") as turn: await runtime.generate_with_skills(call, "S", "U") self.assertFalse(turn.references) self.assertIn("token_budget", call.await_args.args[0]) self.assertNotIn("巨", call.await_args.args[0]) async def test_text_envelope_after_original_and_unwrap(self): call = AsyncMock(side_effect=[text(tool(name="alpha"), content="NEVER SEND"), text(content="最终正文")]) original = "只输出正文，不要JSON" with runtime.skill_session("help", "text"): self.assertEqual(await runtime.generate_with_skills(call, original, "U"), "最终正文") prompt = call.await_args.args[0] self.assertLess(prompt.index(original), prompt.index("Internal output contract")) self.assertNotIn("NEVER SEND", prompt) async def test_text_business_tools_rejected_including_mixed(self): for tools in ([{"name": "mail"}], [tool(name="alpha"), {"name": "mail"}]): call = AsyncMock(return_value=text(*tools)) with runtime.skill_session("notification", "text"): self.assertIsNone(await runtime.generate_with_skills(call, "S", "U")) self.assertEqual(call.await_count, 1) self.assertEqual(self.snapshot.loads, []) async def test_malformed_envelopes_bounded_no_raw_leak(self): for raw in ("PRIVATE RAW TEXT", '{"message":"not a list"}', '{"message":[],"tools":"bad"}'): call = AsyncMock(return_value=raw) with runtime.skill_session("private_chat"): self.assertIsNone(await runtime.generate_with_skills(call, "S", "U")) self.assertEqual(call.await_count, 3) self.assertNotIn(raw, call.await_args.args[0]) self.assertIn("invalid_envelope", call.await_args.args[0]) async def test_none_immediate_no_extra_api_calls(self): for responses in ([None], [chat(tool(name="alpha")), None]): call = AsyncMock(side_effect=responses) with runtime.skill_session("private_chat"): self.assertIsNone(await runtime.generate_with_skills(call, "S", "U")) self.assertEqual(call.await_count, len(responses)) async def test_final_chat_raw_identity_and_missing_tools(self): final = '```json\n{"message":[{"content":"ok"}],"memory":["keep"],"need_reasoning":true}\n```' with runtime.skill_session("private_chat"): self.assertIs(await runtime.generate_with_skills(AsyncMock(return_value=final), "S", "U"), final) async def test_invalid_skill_requests_safe_observations(self): bad = [tool("write", name="alpha", path="PRIVATE-PATH"), {"name": "skill", "args": []}, tool(name="PRIVATE-UNKNOWN"), tool("load", name=["alpha"])] call = AsyncMock(side_effect=[chat(*bad), chat()]) with runtime.skill_session("private_chat"): await runtime.generate_with_skills(call, "S", "U") prompt = call.await_args.args[0] self.assertIn("invalid_request", prompt) self.assertIn("unknown_skill", prompt) self.assertNotIn("PRIVATE-", prompt) async def test_list_uses_full_deterministic_catalog(self): call = AsyncMock(side_effect=[chat(tool("list")), chat()]) with runtime.skill_session("private_chat"): await runtime.generate_with_skills(call, "S", "U") prompts = [item.args[0] for item in call.await_args_list] for prompt in prompts: self.assertEqual(prompt.count(self.snapshot.catalog_text()), 1) self.assertEqual(prompts[0].split("S\n\n")[0], prompts[1].split("S\n\n")[0]) async def test_pinned_snapshot_followups_and_nested_new_sessions(self): self.snapshot.always = ("alpha",) self.snapshot.scenes = {"tool_result": ("beta",)} original = self.snapshot replacement = Snapshot(always=("gamma",)) with runtime.skill_session("private_chat") as outer: self.registry.snapshot.return_value = replacement with runtime.skill_scene("tool_result") as same: self.assertIs(same, outer) self.assertEqual(set(same.loaded), {"alpha", "beta"}) self.assertIs(same.snapshot, original) with runtime.skill_session("help", "text") as inner: self.assertIsNot(inner, outer) self.assertIs(inner.snapshot, replacement) self.assertEqual(set(inner.loaded), {"gamma"}) self.assertIs(runtime.current_turn(), outer) call = AsyncMock(return_value=chat()) await runtime.generate_with_skills(call, "S", "U") self.assertIn('"scene": "private_chat"', call.await_args.args[0]) self.assertIn("EXACT beta body", call.await_args.args[0]) self.assertIsNone(runtime.current_turn()) self.assertEqual(self.registry.snapshot.call_count, 2) async def test_parallel_session_isolation(self): both_entered = asyncio.Event() entered = [] async def worker(name): with runtime.skill_session("private_chat") as turn: turn.load(name) entered.append(name) if len(entered) == 2: both_entered.set() await both_entered.wait() self.assertIs(runtime.current_turn(), turn) call = AsyncMock(return_value=chat()) await runtime.generate_with_skills(call, "S", "U") return set(turn.loaded), call.await_args.args[0] first, second = await asyncio.gather(worker("alpha"), worker("beta")) self.assertEqual(first[0], {"alpha"}) self.assertEqual(second[0], {"beta"}) self.assertNotIn("EXACT beta body", first[1]) self.assertNotIn("EXACT alpha body", second[1]) self.assertIsNone(runtime.current_turn()) async def test_no_implicit_session_suspend_and_exception_restoration(self): with runtime.skill_scene("tool_result") as absent: self.assertIsNone(absent) self.assertEqual(runtime.estimated_overhead(), 0) self.registry.snapshot.assert_not_called() with runtime.skill_session("private_chat") as turn: self.assertEqual(runtime.estimated_overhead(), 12000) with self.assertRaisesRegex(RuntimeError, "expected"): with runtime.suspend_skills(): self.assertIsNone(runtime.current_turn()) call = AsyncMock(return_value="plain") self.assertEqual(await runtime.generate_with_skills(call, "S", "U"), "plain") call.assert_awaited_once_with("S", "U", 0) raise RuntimeError("expected") self.assertIs(runtime.current_turn(), turn) self.assertEqual(runtime.estimated_overhead("private_chat"), 12000) async def test_followup_reuses_round_budget_and_forces_reasoning(self): first = AsyncMock(side_effect=[chat(tool(name="alpha")), chat()]) followup = AsyncMock(side_effect=[chat(tool(name="beta")), chat()]) with runtime.skill_session("private_chat") as turn: await runtime.generate_with_skills(first, "S", "U") with runtime.skill_scene("tool_result"): await runtime.generate_with_skills(followup, "S2", "U2", reasoning=False) self.assertEqual(turn.rounds, 2) self.assertTrue(followup.await_args_list[0].kwargs["reasoning"]) self.assertEqual(followup.await_args.args[0].count("EXACT alpha body"), 1) async def test_always_stable_prefix_scene_and_original_not_truncated(self): self.snapshot.always = ("alpha",) self.snapshot.scenes = {"tool_result": ("beta",)} original = "ORIGINAL-START" + "历史" * 20000 + "ORIGINAL-END" call = AsyncMock(return_value=chat()) with runtime.skill_session("private_chat"): await runtime.generate_with_skills(call, original, "U") with runtime.skill_scene("tool_result"): await runtime.generate_with_skills(call, original, "U") first, second = [item.args[0] for item in call.await_args_list] self.assertIn(original, first) self.assertIn(original, second) self.assertEqual(first.split("[Skill source")[0], second.split("[Skill source")[0]) self.assertLess(second.index("EXACT alpha body"), second.index(original)) self.assertLess(second.index("EXACT beta body"), second.index(original)) async def test_reminder_inherits_notification_binding(self): self.snapshot.scenes = {"notification": ("alpha",)} with runtime.skill_session("reminder", "text") as turn: self.assertEqual(set(turn.loaded), {"alpha"}) async def test_tiny_budget_fails_before_transport(self): self.snapshot.limits["max_tokens"] = 10 call = AsyncMock(return_value=chat()) with runtime.skill_session("private_chat") as turn: self.assertIsNone(await runtime.generate_with_skills(call, "S", "U")) self.assertEqual(turn.observations[-1]["reason"], "token_budget") call.assert_not_awaited() async def test_final_skill_disabled_allows_final_business_plan(self): final = chat({"name": "mail"}) call = AsyncMock(side_effect=[chat(tool(name="alpha")), chat(tool(name="beta")), final]) with runtime.skill_session("private_chat"): self.assertEqual(await runtime.generate_with_skills(call, "S", "U"), final) self.assertEqual(call.await_count, 3) self.assertIn("skill tools disabled", call.await_args.args[0]) async def test_text_malformed_is_not_leaked_and_can_repair(self): call = AsyncMock(side_effect=["PRIVATE RAW ANSWER", text(content="repaired")]) with runtime.skill_session("help", "text"): self.assertEqual(await runtime.generate_with_skills(call, "S", "U"), "repaired") self.assertNotIn("PRIVATE RAW ANSWER", call.await_args.args[0]) self.assertTrue(call.await_args.kwargs["reasoning"]) async def test_load_exposes_reference_names_but_not_unread_source(self): self.snapshot.records["alpha"]["references"]["../private.md"] = "hidden" call = AsyncMock(side_effect=[chat(tool(name="alpha")), chat()]) with runtime.skill_session("private_chat"): await runtime.generate_with_skills(call, "S", "U") prompt = call.await_args.args[0] self.assertIn('"available_references": ["references/note.md"]', prompt) self.assertNotIn("EXACT reference", prompt) self.assertNotIn("../private.md", prompt) self.assertEqual(self.snapshot.reads, []) async def test_reserved_budget_never_forwarded_even_disabled(self): call = AsyncMock(return_value="plain") self.assertEqual(await runtime.generate_with_skills(call, "S", "U", _prompt_token_budget=1), "plain") call.assert_awaited_once_with("S", "U", 0) self.registry.snapshot.assert_not_called() async def test_total_text_budget_rejects_without_cutting_or_transport(self): call = AsyncMock(return_value=chat()) with runtime.skill_session("private_chat") as turn: result = await runtime.generate_with_skills(call, "S" * 4000, "中" * 1000, _prompt_token_budget=1000) self.assertIsNone(result) self.assertEqual(turn.observations[-1]["reason"], "total_token_budget") call.assert_not_awaited() async def test_total_text_budget_uses_engine_compatible_estimate(self): call = AsyncMock(return_value=chat()) original = "a" * 20000 with runtime.skill_session("private_chat"): result = await runtime.generate_with_skills(call, original, "U", _prompt_token_budget=6000) self.assertEqual(result, chat()) self.assertNotIn("_prompt_token_budget", call.await_args.kwargs) self.assertIn(original, call.await_args.args[0]) async def test_small_valid_budget_can_still_load_short_source(self): self.snapshot.limits["max_tokens"] = 1200 call = AsyncMock(side_effect=[chat(tool(name="alpha")), chat()]) with runtime.skill_session("private_chat") as turn: self.assertEqual(await runtime.generate_with_skills(call, "S", "U"), chat()) self.assertIn("alpha", turn.loaded) async def test_summary_prints_safe_metadata_and_bound_priority(self): self.snapshot.always = ("alpha",) with runtime.skill_session("private_chat"): pass summary = self.summary_print.call_args.args[0] self.assertIn("[skills]", summary) self.assertIn("fake-pinned-fingerprint", summary) self.assertIn('"ids": ["alpha"]', summary) self.assertIn('"tokens":', summary) self.assertNotIn("EXACT alpha body", summary) self.assertIn("personal expression preferences override skill", runtime._RULES) async def test_logs_do_not_contain_sources_or_user_data(self): call = AsyncMock(side_effect=[chat(tool(name="alpha"), tool(name="PRIVATE USER SECRET")), chat()]) with self.assertLogs("skill_runtime", "DEBUG") as logs: with runtime.skill_session("private_chat"): await runtime.generate_with_skills(call, "SECRET SYSTEM", "SECRET USER") output = "\n".join(logs.output) self.assertNotIn("SECRET", output) self.assertNotIn("EXACT alpha body", output) self.assertIn("hash=", output) if __name__ == "__main__": unittest.main()