1. SecurityClaw 学习笔记/

06 从技能目录到调度任务:动态技能系统的最小闭环

结论先说:动态技能系统的难点不是“发现技能”,而是“把发现到的技能变成受控能力” #

很多人第一次做技能系统,都会被 importlib.import_module() 吸引,觉得动态加载最难。其实不是。

真正难的是后半段:

  1. 哪些目录算有效技能
  2. 哪些技能允许手动触发,哪些要自动调度
  3. 一个技能出错时,为什么不能拖垮整个系统
  4. 技能的调度元数据该放哪里,运行时又该信任到什么程度

一、最小闭环是什么 #

先给出这篇最重要的模型:

磁盘目录
  -> SkillLoader 发现并导入
  -> Runner 分类与注册
  -> Scheduler 负责时间触发
  -> dispatch 负责手动触发
  -> Skill.run(context) 真正执行

对应伪代码:

class SkillLoader:
    def discover(self, skill_root: str) -> list[Skill]:
        skills = []
        for subdir in os.listdir(skill_root):
            skill = self._load_skill(subdir)
            if skill:
                skills.append(skill)
        return skills


class Runner:
    def setup(self, skills: list[Skill]):
        for skill in skills:
            if skill.interval or skill.cron:
                self.scheduler.register(skill)
            else:
                self.manual_registry[skill.name] = skill

    def dispatch(self, name: str, context: dict):
        return self.manual_registry[name].run(context)

这个闭环的价值在于:发现、分类、调度、执行四步被拆开了


二、加载器的职责:目录是不是一个合法技能 #

SkillLoader 不应该关心业务语义,它只该判断这个目录能不能变成可运行能力。

2.1 最小判定条件 #

从当前项目的设计看,一个目录至少要满足:

  • logic.py
  • logic.py 能被导入
  • 模块里暴露 run
# 伪代码:加载器判断技能是否合法
mod = importlib.import_module(f"skills.{dir_name}.logic")
if not hasattr(mod, "run"):
    return None
return Skill(name=dir_name, run=mod.run, instruction=load_instruction(...))

2.2 为什么要宽容失败 #

如果某个技能因为:

  • 语法错误
  • 缺依赖
  • 没有 run
  • instruction 不完整

就把整个加载过程炸掉,那这套系统根本不适合长期运行。

所以合理的策略是:

一个技能坏了,只跳过这一个技能,不影响其他技能注册。

这不是“放松约束”,而是隔离故障域


三、为什么调度逻辑不应该写进 Loader #

这是这套设计里最值得讲的 tradeoff 之一。

方案 A:Loader 负责加载 + 调度 #

优点:简单,文件少 缺点:

  • 职责混在一起
  • 很难单独测试“调度分类是否正确”
  • 加载器开始依赖时间系统和调度框架

方案 B:Loader 只负责发现,Runner 负责分类,Scheduler 负责时间触发 #

优点:

  • 责任边界清楚
  • 手动技能和自动技能共用同一个 Skill 抽象
  • 测试容易拆开 缺点:
  • 模块数变多
  • 读代码时需要跨三层跳转

SecurityClaw 走的是 B。这是正确的。因为“动态加载”本质是文件系统问题,而“定时执行”本质是运行时编排问题,它们不是一件事。


四、调度元数据应该放在哪 #

当前设计把调度信息放进 instruction.md 或相邻元数据里,而不是写死在代码里。

这件事的意义很大:

4.1 让技能本身携带运行方式 #

一个技能如果带了:

  • interval
  • cron
  • run_on_first_startup

那它就不只是“一个函数”,而是“一个带运行契约的能力单元”。

4.2 让运行时可以统一解释 #

Runner 只看这些声明,不需要知道业务内容:

if skill.interval:
    scheduler.add_interval_job(skill.run, skill.interval)
elif skill.cron:
    scheduler.add_cron_job(skill.run, skill.cron)
else:
    manual_registry[skill.name] = skill

这就是“声明式能力系统”的核心:能力由文件定义,运行方式由元数据声明,框架只负责解释。


五、完整时序:从目录到调度 #

sequenceDiagram
    participant Disk as 技能目录
    participant Loader as SkillLoader
    participant Runner as Runner
    participant Scheduler as Scheduler

    Disk->>Loader: os.listdir(skills/)
    Loader->>Loader: 导入 logic.py / 检查 run
    alt skill 非法
        Loader-->>Runner: 跳过
    else skill 合法
        Loader-->>Runner: 返回 Skill 对象
        Runner->>Runner: 解析 instruction / manifest
        alt 有 interval 或 cron
            Runner->>Scheduler: 注册定时任务
        else 无调度元数据
            Runner->>Runner: 注册为手动技能
        end
    end

这张图里最值得注意的一点是:Scheduler 不需要知道技能来自哪里。它只接收一个已经被 Runner 分类过的“可执行能力”。

这正是分层价值所在。


六、失败路径:动态技能系统最容易炸的地方 #

6.1 空目录 #

应该跳过,不报 fatal

6.2 logic.py 缺失 #

应该跳过,不注册

6.3 run 不存在 #

应该跳过,并留下结构化 warning

6.4 元数据冲突 #

例如同时声明 intervalcron

这时必须有优先级规则,否则运行时语义不确定。

6.5 技能导入阶段连外部依赖 #

如果技能在 import 时就去连数据库或打网络,那 Loader 会变得非常脆弱。

正确做法应该是:

  • import 阶段只加载定义
  • 真正外部依赖放进 run(context)

这是保证“发现”和“执行”分层的关键。


七、测试为什么要围绕 tmp_path 写 #

动态加载最难测的不是业务逻辑,而是目录形态。最好的办法就是:

  • 临时建目录
  • 临时写 logic.py
  • 临时写元数据
  • discover(tmp_path)
  • 验证结果
def test_skip_no_run(tmp_path):
    skill_dir = tmp_path / "bad_skill"
    skill_dir.mkdir()
    (skill_dir / "logic.py").write_text("x = 1")

    skills = loader.discover(tmp_path)
    assert "bad_skill" not in [s.name for s in skills]

这种测试之所以重要,是因为它证明的不是“技能业务对不对”,而是“框架能不能在脏环境下保持稳定”。


八、这一篇要你真正记住的框架 #

动态技能系统不是“会 import 的插件机制”,而是“把目录、元数据、时间调度和手动触发整合成一个可控能力系统”。

你可以用这四层记它:

发现(Loader)
  -> 分类(Runner)
  -> 触发(Scheduler / dispatch)
  -> 执行(Skill.run)

只要这四层混了两层,系统就会开始难测、难扩展、难恢复。


九、验证命令 #

pytest tests/test_skill_loader.py -v --tb=short
pytest tests/test_runner.py tests/test_scheduler.py -v