fix(aibox): round-5 audit batch K - max_turns clamp, task input validation, XSS hardening, atomic writes, broadcast timeout

第五轮审计（J1~J16）全量修复 + 子代理 max_turns 钳制：
- 子代理 max_turns 钳制（0/None 回退默认 10，负数钳制 1）+ 14 项回归测试
- J1 任务系统投毒：add_memo 入库校验（时间/间隔/类型）、调度器预检与触发事务
  防御、展示路径兜底、register 移入 try、add_user_task 补 repeat_end_time 透传
- J2/J13 前端 XSS：三页面 marked 输出经 DOMPurify 净化、entry1 系统消息/任务
  分隔条改 textContent、CDN 脚本补 SRI
- J3 FTS 删除换序（先 fts 后内容表，索引不再残留）
- J5 原子写推广：新增 common/atomic.py（mkstemp+fsync+os.replace），config/prompt/
  history 全部接入；损坏配置备份后回退
- J6 单例初始化加锁（PromptManager/ConfigManager 并发首例不双重加载）
- J7 broadcast 并发发送 + wait_for 超时回收僵死连接
- J4/J8/J9~J12/J14~J16：验证异常脱敏、交互输入容错、MemoDatabase WAL+busy_timeout、
  memo 批量删除、/new 竞态、skills 引号转义、voice_client 清理、死代码移除、
  skill_manager 同名键/help 截断

aibox/tests/test_k_fixes.py

@@ -0,0 +1,540 @@
+"""aibox 第五轮审计（Batch K）修复回归测试（2026-08-19）
+
+波次 1：
+  - J3：FTS 外部内容表删除换序后索引无残留（delete_memory / delete_memories_by_ids）
+  - J4：model_config 连通性验证异常消息 Bearer 令牌脱敏
+  - J6：PromptManager/ConfigManager 单例并发初始化不双重加载
+  - J8：add_profile 对非法温度/超时的容错（交互层不再抢先 float()/int()）
+
+波次 2（J1 任务系统投毒）：
+  - add_memo 入库校验：畸形时间/重复间隔/重复类型拒绝，三入口一处覆盖
+  - 存量脏数据防御：调度器预检、mark_as_triggered、展示路径不再打崩
+  - WS task add 友好错误 + register 握手异常可回收
+"""
+
+import asyncio
+import json
+import os
+import sqlite3
+import sys
+import threading
+import types
+from unittest import mock
+
+import pytest
+
+if "websockets" not in sys.modules:
+    sys.modules["websockets"] = types.ModuleType("websockets")
+
+import aibox  # noqa: E402
+import model_config  # noqa: E402
+from common import ConfigManager  # noqa: E402
+from subsystems.db import MemoDatabase, MemoryDatabase  # noqa: E402
+from subsystems.prompt_manager import PromptManager  # noqa: E402
+
+from test_model_config import _cfg  # noqa: E402
+
+
+# ==================== J3：FTS 删除顺序 ====================
+
+def _fts_hits(db_path, keyword):
+    conn = sqlite3.connect(str(db_path))
+    try:
+        return conn.execute(
+            "SELECT rowid FROM memories_fts WHERE memories_fts MATCH ?",
+            (f'"{keyword}"',)).fetchall()
+    finally:
+        conn.close()
+
+
+def _add_mem(db, content):
+    return db.add_memory(topic="话题", summary="摘要", content=content,
+                         start_time="2026-08-19 00:00:00",
+                         end_time="2026-08-19 00:00:01", importance=2)
+
+
+class TestJ3FtsDeleteOrder:
+    def test_delete_memory_leaves_no_fts_orphan(self, tmp_path):
+        """单条删除后 FTS 索引不得残留（修复前 MATCH 仍命中已删 rowid）"""
+        db = MemoryDatabase(db_path=str(tmp_path / "mem.db"))
+        mid = _add_mem(db, "机密内容甲")
+        assert _fts_hits(tmp_path / "mem.db", "机密内容甲")
+        assert db.delete_memory(mid) is True
+        assert _fts_hits(tmp_path / "mem.db", "机密内容甲") == []
+
+    def test_batch_delete_leaves_no_fts_orphan(self, tmp_path):
+        """批量删除同样不留索引残留"""
+        db = MemoryDatabase(db_path=str(tmp_path / "mem.db"))
+        ids = [_add_mem(db, f"机密内容乙{i}") for i in range(3)]
+        assert db.delete_memories_by_ids(ids) == 3
+        for i in range(3):
+            assert _fts_hits(tmp_path / "mem.db", f"机密内容乙{i}") == []
+
+    def test_search_after_delete_no_hit(self, tmp_path):
+        """删除后搜索（FTS 路径）不再命中已删内容"""
+        db = MemoryDatabase(db_path=str(tmp_path / "mem.db"))
+        mid = _add_mem(db, "独特关键词丙")
+        assert db.search_memories(["独特关键词丙"])
+        db.delete_memory(mid)
+        assert db.search_memories(["独特关键词丙"]) == []
+
+
+# ==================== J4：验证异常脱敏 ====================
+
+class TestJ4TestProfileSanitize:
+    def _models_cfg(self, tmp_path):
+        return _cfg(tmp_path, {"models": {"bad": {
+            "api_url": "https://api.example.com/v1/chat/completions",
+            "model": "m", "api_key_env": "K"}}})
+
+    def test_bearer_token_masked_in_failure_message(self, tmp_path):
+        """网关错误页回显 Authorization 头时，令牌不得出现在终端回显"""
+        cm = self._models_cfg(tmp_path)
+        err = RuntimeError("HTTP 502: Authorization: Bearer sk-secret12345")
+        with mock.patch("llm_client.create_llm_client", side_effect=err):
+            result = model_config.test_profile(cm, "bad")
+        assert "sk-secret12345" not in result
+        assert "Bearer ***" in result
+
+    def test_plain_error_message_preserved(self, tmp_path):
+        """无敏感内容的错误消息原样保留（脱敏不误伤）"""
+        cm = self._models_cfg(tmp_path)
+        with mock.patch("llm_client.create_llm_client",
+                        side_effect=RuntimeError("连接被拒绝")):
+            result = model_config.test_profile(cm, "bad")
+        assert "连接被拒绝" in result
+
+
+# ==================== J6：单例并发初始化 ====================
+
+class TestJ6SingletonInitRace:
+    def test_prompt_manager_concurrent_init_no_duplication(self, tmp_path):
+        """多线程首例并发：单实例且提示词只加载一份（修复前整份翻倍）"""
+        PromptManager._instance = None
+        PromptManager._prompts = []
+        PromptManager._prompts_dir = None
+        prompts_dir = tmp_path / "prompts"
+        prompts_dir.mkdir()
+        try:
+            barrier = threading.Barrier(8)
+            instances = [None] * 8
+
+            def worker(k):
+                barrier.wait()
+                instances[k] = PromptManager()
+
+            with mock.patch.object(ConfigManager, "get",
+                                   return_value=str(prompts_dir)):
+                threads = [threading.Thread(target=worker, args=(k,))
+                           for k in range(8)]
+                for t in threads:
+                    t.start()
+                for t in threads:
+                    t.join()
+
+            assert len({id(i) for i in instances}) == 1
+            files = [p["file"] for p in PromptManager._prompts]
+            assert files, "提示词应已加载"
+            assert len(files) == len(set(files)), f"提示词被重复加载: {files}"
+        finally:
+            PromptManager._instance = None
+            PromptManager._prompts = []
+            PromptManager._prompts_dir = None
+
+    def test_config_manager_concurrent_init_single_load(self, tmp_path):
+        """多线程首例并发：单实例且 _load_config 只执行一次"""
+        cfg_file = tmp_path / "config.json"
+        calls = {"n": 0}
+        original_load = ConfigManager._load_config
+
+        def counting_load(self):
+            calls["n"] += 1
+            original_load(self)
+
+        ConfigManager._instance = None
+        ConfigManager._config = None
+        ConfigManager._config_path = None
+        try:
+            barrier = threading.Barrier(8)
+            instances = [None] * 8
+
+            def worker(k):
+                barrier.wait()
+                instances[k] = ConfigManager()
+
+            with mock.patch.dict(os.environ, {"AI_CONFIG_PATH": str(cfg_file)}), \
+                 mock.patch.object(ConfigManager, "_load_config", counting_load):
+                threads = [threading.Thread(target=worker, args=(k,))
+                           for k in range(8)]
+                for t in threads:
+                    t.start()
+                for t in threads:
+                    t.join()
+
+            assert len({id(i) for i in instances}) == 1
+            assert calls["n"] == 1
+        finally:
+            ConfigManager._instance = None
+            ConfigManager._config = None
+            ConfigManager._config_path = None
+
+
+# ==================== J8：交互输入容错 ====================
+
+class TestJ8NumericTolerance:
+    def test_invalid_temperature_returns_friendly_error(self, tmp_path):
+        """非数字温度 → 友好错误而非 ValueError（交互层不再抢先 float()）"""
+        cm = _cfg(tmp_path, {"models": {}})
+        msgs = model_config.add_profile(
+            cm, "p1", "https://api.example.com/v1/chat/completions", "m1",
+            api_key_env="K", temperature="not-a-number")
+        assert any(m.startswith("❌") and "温度" in m for m in msgs)
+        assert "p1" not in model_config.get_models(cm)
+
+    def test_invalid_timeout_returns_friendly_error(self, tmp_path):
+        cm = _cfg(tmp_path, {"models": {}})
+        msgs = model_config.add_profile(
+            cm, "p1", "https://api.example.com/v1/chat/completions", "m1",
+            api_key_env="K", timeout="abc")
+        assert any(m.startswith("❌") and "超时" in m for m in msgs)
+        assert "p1" not in model_config.get_models(cm)
+
+    def test_string_numbers_accepted_and_clamped(self, tmp_path):
+        """数字字符串正常接受并钳制（交互层原始字符串直传语义）"""
+        cm = _cfg(tmp_path, {"models": {}})
+        msgs = model_config.add_profile(
+            cm, "p1", "https://api.example.com/v1/chat/completions", "m1",
+            api_key_env="K", temperature="0.7", timeout="120")
+        assert all(not m.startswith("❌") for m in msgs)
+        profile = model_config.get_models(cm)["p1"]
+        assert profile["temperature"] == 0.7
+        assert profile["timeout"] == 120
+
+
+# ==================== J1 波次 2：任务系统投毒 ====================
+
+def _insert_memo_raw(db_path, reminder_time="2026-08-18 00:00:00",
+                     repeat_type="none", repeat_interval=0,
+                     repeat_interval_unit="days", repeat_end_time=None):
+    """直接 SQL 插入任务行——模拟校验落地前的存量脏数据（绕过 add_memo）"""
+    with sqlite3.connect(str(db_path)) as conn:
+        cur = conn.cursor()
+        cur.execute('''INSERT INTO memos
+            (title, content, created_at, reminder_time, tags, priority, created_by,
+             is_completed, source, repeat_type, repeat_interval, repeat_interval_unit,
+             repeat_end_time, is_immediate)
+            VALUES ('t', 'c', '2026-08-18 00:00:00', ?, '', 1, 'test',
+             0, 'test', ?, ?, ?, ?, 0)''',
+            (reminder_time, repeat_type, repeat_interval, repeat_interval_unit,
+             repeat_end_time))
+        return cur.lastrowid
+
+
+class TestJ1AddMemoValidation:
+    def test_malformed_reminder_time_rejected(self, tmp_path):
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        with pytest.raises(ValueError, match="提醒时间格式非法"):
+            db.add_memo(title="t", reminder_time="不是时间")
+        with sqlite3.connect(str(tmp_path / "m.db")) as conn:
+            assert conn.execute("SELECT COUNT(*) FROM memos").fetchone()[0] == 0
+
+    def test_malformed_repeat_end_time_rejected(self, tmp_path):
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        with pytest.raises(ValueError, match="重复结束时间格式非法"):
+            db.add_memo(title="t", reminder_time="2026-08-18 00:00:00",
+                        repeat_type="daily", repeat_end_time="不是时间")
+
+    def test_invalid_repeat_type_rejected(self, tmp_path):
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        with pytest.raises(ValueError, match="无效的重复类型"):
+            db.add_memo(title="t", reminder_time="2026-08-18 00:00:00",
+                        repeat_type="every-second")
+
+    def test_non_numeric_repeat_interval_rejected(self, tmp_path):
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        with pytest.raises(ValueError, match="重复间隔必须是整数"):
+            db.add_memo(title="t", reminder_time="2026-08-18 00:00:00",
+                        repeat_type="custom", repeat_interval="垃圾")
+
+    def test_custom_repeat_requires_positive_interval(self, tmp_path):
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        with pytest.raises(ValueError, match="repeat_interval"):
+            db.add_memo(title="t", reminder_time="2026-08-18 00:00:00",
+                        repeat_type="custom", repeat_interval=0)
+
+    def test_numeric_string_interval_coerced(self, tmp_path):
+        """数字字符串被强转接受（WS 传 JSON 数字/字符串均安全）"""
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        mid = db.add_memo(title="t", reminder_time="2026-08-18 00:00:00",
+                          repeat_type="custom", repeat_interval="30",
+                          repeat_interval_unit="minutes")
+        assert db.get_memo(mid)["repeat_interval"] == 30
+
+    def test_add_user_task_propagates_validation(self, tmp_path):
+        """WS 路径（add_user_task 薄包装 add_memo）同样被校验拦截"""
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        with pytest.raises(ValueError, match="提醒时间格式非法"):
+            db.add_user_task(title="t", reminder_time="不是时间")
+
+
+class TestJ1DirtyDataDefense:
+    def test_mark_as_triggered_completes_malformed_one_shot(self, tmp_path):
+        """存量畸形 reminder_time 一次性任务：触发不再抛异常，正常完成"""
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        mid = _insert_memo_raw(tmp_path / "m.db", reminder_time="不是时间")
+        assert db.mark_as_triggered(mid) is None
+        assert db.get_memo(mid)["is_completed"] == 1
+
+    def test_mark_as_triggered_handles_garbage_repeat_interval(self, tmp_path):
+        """存量 custom 任务 repeat_interval 为文本：按非重复处理，不打崩事务"""
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        mid = _insert_memo_raw(tmp_path / "m.db", reminder_time="2026-08-18 00:00:00",
+                               repeat_type="custom", repeat_interval="垃圾")
+        assert db.mark_as_triggered(mid) is None
+        assert db.get_memo(mid)["is_completed"] == 1
+
+    def test_malformed_repeat_end_time_treated_as_no_end(self, tmp_path):
+        """存量畸形 repeat_end_time：解析失败视为无结束时间，不抛异常"""
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        mid = _insert_memo_raw(tmp_path / "m.db", reminder_time="2026-08-19 00:00:00",
+                               repeat_type="daily", repeat_end_time="不是时间")
+        memo = db.get_memo(mid)
+        from datetime import datetime
+        base = datetime.strptime("2026-08-19 00:00:00", "%Y-%m-%d %H:%M:%S")
+        assert db._should_end_repeat(memo, base) is False
+        assert db._calculate_next_reminder(memo, base) is not None
+
+    def test_pending_tasks_for_user_malformed_time_overdue(self, tmp_path):
+        """展示路径：畸形 reminder_time 行按到期展示，不再打崩握手"""
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        mid = _insert_memo_raw(tmp_path / "m.db", reminder_time="不是时间")
+        tasks = db.get_pending_tasks_for_user()
+        row = next(t for t in tasks if t["id"] == mid)
+        assert row["is_overdue"] is True
+
+
+class TestJ1WsAndScheduler:
+    def test_ws_task_add_rejects_malformed_repeat_end(self, tmp_path):
+        """WS task add 传畸形 repeat_end_time：友好错误且不入库"""
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        ai_manager = mock.MagicMock()
+        ai_manager.memo_db = db
+        handler = aibox.WebSocketHandler(ai_manager)
+        try:
+            ws = mock.AsyncMock()
+            asyncio.run(handler.handle_task_message(ws, {
+                "type": "task", "action": "add", "title": "t", "delay": 5,
+                "repeat_type": "daily", "repeat_end_time": "不是时间"}))
+            sent = json.loads(ws.send.call_args[0][0])
+            assert sent["type"] == "error"
+            assert "任务添加失败" in sent["content"]
+            with sqlite3.connect(str(tmp_path / "m.db")) as conn:
+                assert conn.execute("SELECT COUNT(*) FROM memos").fetchone()[0] == 0
+        finally:
+            handler.shutdown()
+
+    def test_scheduler_precheck_malformed_time_executes_not_loops(self, tmp_path):
+        """调度器预检：畸形 reminder_time 按到期执行并标记触发（不再热循环）"""
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        mid = _insert_memo_raw(tmp_path / "m.db", reminder_time="不是时间")
+        loop = asyncio.new_event_loop()
+        executed = {"n": 0}
+        try:
+            scheduler = aibox.TaskScheduler(loop=loop, logger=None)
+            scheduler.broadcast_enabled = False
+
+            async def fake_execute(memo, memo_db):
+                executed["n"] += 1
+                return "ok"
+
+            scheduler.execute_task_as_conversation = fake_execute
+            loop.run_until_complete(scheduler.check_and_execute_tasks(db))
+        finally:
+            loop.close()
+        assert executed["n"] == 1
+        assert db.get_memo(mid)["is_completed"] == 1
+
+
+# ==================== J5 波次 4：原子写 + 损坏备份 ====================
+
+class TestJ5AtomicWrites:
+    def test_save_config_atomic_no_tmp_leftover(self, tmp_path):
+        """_save_config 原子写：无 .aibox_tmp_ 残留且 JSON 合法"""
+        cm = ConfigManager()
+        cm._config_path = str(tmp_path / "config.json")
+        cm._config = {"api": {"deepseek_model": "m"}, "models": {}}
+        assert cm._save_config() is True
+        import json as _json
+        data = _json.loads((tmp_path / "config.json").read_text(encoding="utf-8"))
+        assert data["api"]["deepseek_model"] == "m"
+        assert list(tmp_path.glob(".aibox_tmp_*")) == []
+
+    def test_corrupted_config_backed_up_before_fallback(self, tmp_path):
+        """损坏配置加载：备份 .corrupt.* 文件后回退默认，原内容不丢"""
+        cfg_file = tmp_path / "config.json"
+        cfg_file.write_text('{"api": {"broken": ', encoding="utf-8")  # 半截 JSON
+        ConfigManager._instance = None
+        ConfigManager._config = None
+        ConfigManager._config_path = None
+        try:
+            with mock.patch.dict(os.environ, {"AI_CONFIG_PATH": str(cfg_file)}):
+                cm = ConfigManager()
+            assert cm._config is not None
+            backups = list(tmp_path.glob("config.json.corrupt.*"))
+            assert len(backups) == 1
+            assert backups[0].read_text(encoding="utf-8") == '{"api": {"broken": '
+        finally:
+            ConfigManager._instance = None
+            ConfigManager._config = None
+            ConfigManager._config_path = None
+
+    def test_prompt_write_atomic(self, tmp_path):
+        """prompt 更新原子写：内容生效且无临时文件残留"""
+        from subsystems.prompt_manager import PromptManager
+        PromptManager._instance = None
+        PromptManager._prompts = []
+        PromptManager._prompts_dir = None
+        prompts_dir = tmp_path / "prompts"
+        prompts_dir.mkdir()
+        try:
+            with mock.patch.object(ConfigManager, "get",
+                                   return_value=str(prompts_dir)):
+                pm = PromptManager()
+                assert pm.update_identity("新身份内容") is True
+                identity = prompts_dir / "00_IDENTITY.md"
+                assert "新身份内容" in identity.read_text(encoding="utf-8")
+            assert list(tmp_path.glob(".aibox_tmp_*")) == []
+        finally:
+            PromptManager._instance = None
+            PromptManager._prompts = []
+            PromptManager._prompts_dir = None
+
+
+# ==================== J7 波次 4：broadcast 超时回收 ====================
+
+class TestJ7BroadcastTimeout:
+    def test_stuck_client_timed_out_and_unregistered(self):
+        """半开 TCP 僵死客户端：broadcast 在超时内返回并回收该连接"""
+        from aibox import WebSocketHandler
+        handler = WebSocketHandler(ai_manager=mock.MagicMock())
+        handler.broadcast_send_timeout = 0.1
+        try:
+            ws = mock.AsyncMock()
+
+            async def stuck_send(*args, **kwargs):
+                await asyncio.sleep(30)
+
+            ws.send.side_effect = stuck_send
+            handler.connected_clients.add(ws)
+            asyncio.run(handler.broadcast({"type": "ping"}))
+            assert ws not in handler.connected_clients
+        finally:
+            handler.shutdown()
+
+    def test_healthy_client_stays_connected(self):
+        """正常客户端：发送成功且不被回收"""
+        from aibox import WebSocketHandler
+        handler = WebSocketHandler(ai_manager=mock.MagicMock())
+        handler.broadcast_send_timeout = 1.0
+        try:
+            ws = mock.AsyncMock()
+            handler.connected_clients.add(ws)
+            asyncio.run(handler.broadcast({"type": "ping"}))
+            assert ws in handler.connected_clients
+            assert ws.send.called
+        finally:
+            handler.shutdown()
+
+
+# ==================== J9 波次 5：MemoDatabase 连接层 ====================
+
+class TestJ9MemoDbConnect:
+    def test_wal_enabled_after_init(self, tmp_path):
+        """WAL 显式设置——不再依赖与 MemoryDatabase 的创建顺序"""
+        MemoDatabase(db_path=str(tmp_path / "m.db"))
+        with sqlite3.connect(str(tmp_path / "m.db")) as conn:
+            mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
+        assert mode == "wal"
+
+    def test_connection_factory_sets_busy_timeout(self, tmp_path):
+        """统一连接工厂：busy_timeout 覆盖全部写路径"""
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        conn = db._connect()
+        try:
+            assert conn.execute("PRAGMA busy_timeout").fetchone()[0] == 10000
+        finally:
+            conn.close()
+
+
+# ==================== J10 波次 5：memo 批量删除 ====================
+
+class TestJ10BatchDelete:
+    def test_batch_delete_via_memo_tool(self, tmp_path):
+        """schema 声明的 memo_ids 批量删除真正生效（此前静默忽略）"""
+        from tools.db_tools import MemoTool
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        ids = [db.add_memo(title=f"任务{i}",
+                           reminder_time="2026-08-18 00:00:00") for i in range(3)]
+        tool = MemoTool(db)
+        r = tool.execute(action="delete", confirm=True, memo_ids=ids)
+        assert "3/3" in r
+        with sqlite3.connect(str(tmp_path / "m.db")) as conn:
+            assert conn.execute("SELECT COUNT(*) FROM memos").fetchone()[0] == 0
+
+    def test_batch_delete_requires_list(self, tmp_path):
+        from tools.db_tools import MemoTool
+        db = MemoDatabase(db_path=str(tmp_path / "m.db"))
+        tool = MemoTool(db)
+        r = tool.execute(action="delete", confirm=True, memo_ids="1")
+        assert "列表" in r
+
+
+# ==================== J15 波次 5：未知 task action ====================
+
+class TestJ15UnknownTaskAction:
+    def test_unknown_action_returns_error(self):
+        """未知 task action 明确报错（此前静默吞掉）"""
+        from aibox import WebSocketHandler
+        handler = WebSocketHandler(ai_manager=mock.MagicMock())
+        try:
+            ws = mock.AsyncMock()
+            asyncio.run(handler.handle_task_message(ws, {"action": "explode"}))
+            sent = json.loads(ws.send.call_args[0][0])
+            assert sent["type"] == "error"
+            assert "未知的任务操作" in sent["content"]
+        finally:
+            handler.shutdown()
+
+
+# ==================== J16 波次 5：skill_manager ====================
+
+class TestJ16SkillManager:
+    def test_markdown_skipped_when_executable_same_name(self, tmp_path, monkeypatch):
+        """同名键冲突：可执行技能确定性优先（此前由 listdir 顺序决定）"""
+        from common import constants
+        from subsystems.skill_manager import SkillManager
+        skills_dir = tmp_path / "skills"
+        skills_dir.mkdir()
+        exe = skills_dir / "foo"
+        exe.write_text("#!/bin/sh\necho '可执行技能简介'", encoding="utf-8")
+        exe.chmod(0o755)
+        (skills_dir / "foo.md").write_text("# Markdown 简介\n内容", encoding="utf-8")
+        monkeypatch.setattr(type(constants), "SKILLS_DIR",
+                            property(lambda self: str(skills_dir)))
+        sm = SkillManager()
+        assert "foo" in sm.skills
+        assert sm.skills["foo"]["is_executable"] is True
+
+    def test_help_output_truncated(self, tmp_path, monkeypatch):
+        """--help 输出截断（内容进系统提示词，防撑爆上下文）"""
+        from common import constants
+        from subsystems.skill_manager import SkillManager
+        skills_dir = tmp_path / "skills"
+        skills_dir.mkdir()
+        exe = skills_dir / "big"
+        exe.write_text("#!/bin/sh\necho '" + "x" * 5000 + "'", encoding="utf-8")
+        exe.chmod(0o755)
+        monkeypatch.setattr(type(constants), "SKILLS_DIR",
+                            property(lambda self: str(skills_dir)))
+        sm = SkillManager()
+        assert len(sm.skills["big"]["content"]) <= SkillManager.MAX_HELP_LENGTH + 20