feat: 统一业务扩展注册与启动校验

自动发现 extensions 下的业务模块，集中声明工具、子 Agent、提示词及审批要求。迁移内置能力，支持工具别名与异步工具，统一主子 Agent 审批规则，并更新扩展文档和发布白名单。

验证：211 项测试、结构自检及发布包构建校验通过。

.env.example

@@ -66,6 +66,7 @@ CTX_RECENT_LIMIT=30
 # 人机协同（HITL）
 # ============================================================
 # 需要人工审批的工具（逗号分隔），会追加到默认的计划审批工具之上。
+# 未注册的工具名会阻止启动，避免审批配置拼写错误被忽略。
 # 例：HITL_TOOLS=calculate  → 每次计算都要用户点头，便于自测中断闭环
 HITL_TOOLS=
 

README.md

@@ -130,7 +130,9 @@ DEEP_AGENT_ENV_PRIORITY=process LOG_LEVEL=DEBUG ~/vent-agent/current/release/ser
 
 **目前没有自动的数据迁移、兼容性检查或备份。** 回退只切换代码和依赖，不回滚数据库或 Agent checkpoint；结构不兼容时不能直接切回旧版。完整命令见 [升级与回退](release/README.md#升级与回退)。
 
-## 7. 开发验证与参考
+## 7. 业务扩展与开发验证
+
+新增业务时，在 `extensions/<业务名>/extension.py` 编写 `setup(registry)`，统一声明工具、子 Agent、提示词和审批要求。启动自动发现并校验，无需修改工厂函数；修改后重启生效。示例和完整步骤见 [扩展方式](docs/DEVELOPMENT.md#扩展方式)。
 
 ```bash
 .venv/bin/python -m pytest            # 离线测试，使用临时数据库

agents/agent_factory.py

@@ -20,7 +20,7 @@ from agents import agent_memory, model_factory, prompts
 from agents.middleware import DynamicTimeMiddleware
 from obs import get_logger
 from tools import registry as tool_registry
-from tools.demo_tools import PLAN_APPROVAL_TOOL
+from extensions.registry import approval_names, load_extensions
 
 logger = get_logger(__name__)
 
@@ -128,59 +128,13 @@ def _build_middleware() -> list[Any]:
 
 
 def _build_interrupt_on() -> dict[str, Any]:
-    """需要人工审批的工具集合。
-
-    默认只有计划审批工具。``HITL_TOOLS`` 可追加（逗号分隔的工具名），
-    例如 ``HITL_TOOLS=calculate`` 会要求每次计算都经用户点头——
-    用来演示与自测中断闭环。
-    """
-    names = {PLAN_APPROVAL_TOOL}
-    names.update(name for name in tool_registry.get_tool_names()
-                 if tool_registry.get_spec(name).requires_approval)
-    extra = config.get_str("HITL_TOOLS", "")
-    if extra:
-        names.update(n.strip() for n in extra.split(",") if n.strip())
-
-    available = set(tool_registry.get_tool_names())
-    missing = names - available
-    if missing:
-        logger.warning("interrupt_on 里配置了未注册的工具，已忽略：%s", sorted(missing))
-    return {name: True for name in sorted(names & available)}
+    """审批来自工具声明、扩展声明与 HITL_TOOLS；未知工具在启动时拒绝。"""
+    return {name: True for name in sorted(approval_names())}
 
 
 def _build_subagents() -> list[dict[str, Any]]:
-    """子智能体定义。
-
-    骨架给一个示例子智能体，演示「主 agent 只拿结论、原始材料留在子线程」
-    的用法。子智能体的 ``system_prompt`` **不继承**父级，必须自己写全。
-
-    ``tools`` 必须是 ``BaseTool`` 列表（不是裸函数名，也不能是裸函数）——
-    这里用 ``registry.get_tools_by_name`` 转换，由注册表保证名字有效。
-    """
-    return [
-        {
-            "name": "demo-searcher",
-            "description": (
-                "资料检索专员。当任务需要在演示知识库里做多轮检索、比对多个来源、"
-                "或需要把检索结果整理成摘要时委派给它。"
-                "它只回结论与来源，不回原始文本。"
-            ),
-            "system_prompt": (
-                "你是资料检索专员。职责是在演示知识库里检索并把结果整理成结论。\n\n"
-                "工作方式：\n"
-                "- 用 search_demo_knowledge 检索，可换多组关键词多轮检索。\n"
-                "- 需要数值计算时用 calculate，不要心算。\n"
-                "- 检索不到就如实说「未检索到」，不要用常识补写。\n\n"
-                "输出格式：\n"
-                "- 先给一段不超过 5 句的结论。\n"
-                "- 再列来源编号与标题。\n"
-                "- 不要把检索到的原文整段复制回来，主线程不需要原始数据。"
-            ),
-            "tools": tool_registry.get_tools_by_name(
-                ["search_demo_knowledge", "calculate", "get_current_time"]
-            ),
-        }
-    ]
+    """从统一扩展目录组装子 Agent、系统提示词与工具审批。"""
+    return load_extensions().subagents(approval_names())
 
 
 def build_agent() -> Any:
@@ -220,6 +174,7 @@ def describe() -> dict[str, Any]:
     """组装配置摘要，供 /health 与调试使用。"""
     return {
         "name": "deep-agent",
+        "extensions": list(load_extensions().modules),
         "model": model_factory.describe(),
         "tools": tool_registry.describe(),
         "interrupt_on": sorted(_build_interrupt_on().keys()),

docs/DEVELOPMENT.md

@@ -161,58 +161,82 @@ HTTP 错误语义：额度超限 429；并发超限 429（带 `retry_after`）
 
 ## 扩展方式
 
-### 加一个业务工具
+### 新增业务 Agent
+
+在 `extensions/<业务名>/` 下添加 `__init__.py` 和 `extension.py`。启动按目录名排序，自动加载每个 `extension.py` 的 `setup(registry)`，不需要修改工具模块列表、`agent_factory.py` 或主提示词。
+
+例如 `extensions/ventilation/extension.py`：
 
 ```python
-# tools/my_tools.py
-from tools.registry import register, source, sourced
+from extensions import define_tool
 
-@register(group="我的分组", label="中文显示名", summary="一句话说明")
-def my_tool(query: str, limit: int = 5) -> dict:
-    """做某事。
 
-    Args:
-        query: 查询词。
-        limit: 最多返回几条。
+@define_tool(group="通风", label="计算风量", name="calculate_airflow")
+def airflow(area: float, speed: float) -> float:
+    """根据断面积和平均风速计算风量，返回立方米每秒。
 
-    Returns:
-        含 data 与 sources 的结构；sources 会被渲染成引用标注。
+    Args:
+        area: 断面积，单位平方米，必须大于零。
+        speed: 平均风速，单位米每秒，不能为负。
     """
-    return sourced(
-        data=[...],
-        sources=[source("标题", "https://...", "摘要", score=0.9)],
+    if area <= 0 or speed < 0:
+        raise ValueError("断面积必须大于零，风速不能为负")
+    return area * speed
+
+
+def setup(registry):
+    registry.register_agent(
+        name="ventilation",
+        description="处理矿井通风量计算；需要用断面积和风速求风量时委派。",
+        prompt="你是通风计算助手。先确认单位和输入，调用工具计算，说明结果单位。",
+        tools=[airflow],
     )
 ```
 
-然后把模块名加进 `tools/registry.py` 的 `_BUILTIN_MODULES`。工具名全局唯一，
-重名会在启动时抛错。
+`define_tool` 只声明元数据，不在导入时注册，函数可以直接单测。`setup()` 收集声明，全部通过校验后统一生效；不要在其中访问外部服务或执行业务操作。
 
-**docstring 是给模型看的**：第一行说清做什么，正文说清何时用、参数含义、
-返回什么。模型选错工具多半是因为 docstring 写得含糊。
+- `description` 自动加入主 Agent 的委派说明；`prompt` 是子 Agent 的系统提示词，不继承父级提示词。
+- `tools` 支持函数或已注册工具名。函数自动注册；字符串引用在所有扩展加载完成后校验，允许跨模块引用。
+- 所有注册工具都会提供给主 Agent；子 Agent 只绑定自己的工具列表及框架内置能力。
+- 只加工具时，在 `setup()` 调用 `registry.register_tool(airflow)`。已有工具传给多个 Agent 时会复用同一个注册名。
+- `name` 为实际工具调用名，可以与 Python 函数名不同；异步工具使用 `async def`。
+- 提示词也可以放在业务目录的 Markdown 文件中，通过 `Path(__file__).with_name("prompt.md").read_text(encoding="utf-8")` 读取；发布包会包含这些文件。
 
-### 加一个技能
+内置示例见 `extensions/demo/extension.py`；`extensions/core/extension.py` 注册计划审批和询问用户工具，主提示词与恢复流程依赖它们，需保留。
 
-在 `skills/<name>/SKILL.md` 写带 frontmatter 的操作手册：
+### 声明审批
 
-```markdown
----
-name: my-skill
-description: 什么时候该用这个技能（这段会被模型看到，要写具体）
----
+有写入或外部副作用的工具使用 `@define_tool(..., requires_approval=True)`。也可以在 Agent 定义中设置 `approval_tools=[工具函数或名称]`，但必须包含在该 Agent 的 `tools` 中。
+
+**审批要求按工具合并：** 任意扩展要求某个工具审批后，主 Agent 和所有使用该工具的子 Agent 都会先等待批准。`HITL_TOOLS` 环境变量可以追加工具名；拼写错误会阻止启动。
 
-# 标题
-## 步骤
-1. ...
+### 校验与生效
+
+启动会在数据库初始化、调度器启动之前检查：
+
+- 工具和 Agent 名称格式、重名、与框架内置名称冲突。
+- 工具函数能否生成参数 schema，是否提供 docstring 或 summary。
+- Agent 描述和提示词非空，工具引用有效，审批工具确实绑定。
+- `setup()` 缺失或加载失败、目录/入口符号链接等问题。
+
+新增或修改扩展后**重启服务**，再用 `GET /api/model` 查看 `agent.extensions` 和子 Agent 列表，用 `GET /api/model/tools` 查看工具。`/api/model/reload` 只重建已加载定义，不重新导入业务代码。
+
+开发时运行：
+
+```bash
+.venv/bin/python -m pytest
+.venv/bin/python scripts/selfcheck.py
 ```
 
-agent 启动时只读 frontmatter，按需才 `read_file` 读全文（渐进披露）。
-用户用 `#skill:my-skill` 前缀显式指定。
+自检使用当前数据库配置，建议在开发环境执行。扩展代码及 `.md/.txt/.json/.yaml/.yml` 资源自动进入发布白名单；其他资源类型需调整 `release/build.py`。
+
+旧 `tools.registry.register` 保留兼容；新业务统一使用 `define_tool` 与 `setup(registry)`，避免导入时修改全局注册状态。
+
+### 来源标注和技能
 
-### 加一个子智能体
+工具需要返回引用来源时，使用 `tools.registry.sourced(data=..., sources=[source(...)])`，SSE 层自动生成引用事件。
 
-改 `agents/agent_factory.py` 的 `_build_subagents()`。注意 `tools` 必须是
-`BaseTool` 列表（用 `registry.get_tools_by_name(...)`），`system_prompt`
-不继承父级。
+技能仍放在 `skills/<name>/SKILL.md`，带有 `name`、`description` 的 YAML frontmatter。用户可以用 `#skill:<name>` 指定技能；技能是操作手册，不会替代工具注册。
 
 ---
 
@@ -268,7 +292,8 @@ api/                       chat · sse_core · session · model · usage · sche
   context_meter.py         上下文计量（真实 usage 优先）
   source_tracker.py        来源标注抽取与聚合
 agents/                    agent_factory · agent_registry · model_factory · prompts · middleware · agent_memory
-tools/                     registry（工具唯一事实源）· demo_tools（示例工具）
+extensions/                registry · core/extension.py · demo/extension.py（统一业务扩展）
+tools/                     registry（工具元数据与转换）· demo_tools（示例工具）
 skills/demo-skill/         示例技能
 db/                        db_conn（双方言）· mysql_schema.sql · schema_sqlite.sql · 三个 store
 billing/                   pricing · guard（预检）· meter（入账）
@@ -299,7 +324,7 @@ scripts/                   selfcheck.py · http_e2e.py · start_server.sh
 - `request_plan_approval(plan)`：提交计划等待确认。plan 模式提示词会准确指向这个工具。
 - 批量工具审批卡完整展示每项工具及参数，批准按钮明确写出批次数量。
 - 审批卡带 `interrupt_token`；恢复接口必须原样携带。服务端在会话锁内核对当前中断，过期、缺失和重复确认返回 409。
-- 业务工具可在 `@register(..., requires_approval=True)` 声明强制审批，组装时自动进入框架 HITL。plan 提示词本身不构成所有工具的硬权限边界，真实写操作应声明强制审批。
+- 业务工具可在 `@define_tool(..., requires_approval=True)` 声明强制审批，组装时自动进入框架 HITL。plan 提示词本身不构成所有工具的硬权限边界，真实写操作应声明强制审批。
 - 子智能体和摘要模型的用量计入总费用，其内部正文不混入主回答，也不覆盖主线程上下文占用。
 
 ### 升级已有部署

extensions/__init__.py

@@ -0,0 +1,5 @@
+"""业务扩展入口：每个子目录用 extension.py 的 setup(registry) 声明能力。"""
+from extensions.registry import ExtensionRegistry, load_extensions
+from tools.registry import define_tool
+
+__all__ = ["ExtensionRegistry", "define_tool", "load_extensions"]

extensions/core/__init__.py

@@ -0,0 +1 @@
+"""框架交互能力。"""

extensions/core/extension.py

@@ -0,0 +1,7 @@
+"""主提示词和审批恢复所需的系统工具。"""
+from tools.demo_tools import ask_user, request_plan_approval
+
+
+def setup(registry):
+    registry.register_tool(request_plan_approval)
+    registry.register_tool(ask_user)

extensions/demo/__init__.py

@@ -0,0 +1 @@
+"""内置演示扩展。"""

extensions/demo/extension.py

@@ -0,0 +1,21 @@
+"""演示业务扩展：工具、委派描述和子 Agent 提示词集中声明。"""
+from tools.demo_tools import calculate, get_current_time, search_demo_knowledge
+
+
+def setup(registry):
+    registry.register_agent(
+        name="demo-searcher",
+        description="资料检索专员。当任务需要在演示知识库里做多轮检索、比对多个来源、或整理摘要时委派给它。它只回结论与来源，不回原始文本。",
+        prompt=(
+            "你是资料检索专员。职责是在演示知识库里检索并把结果整理成结论。\n\n"
+            "工作方式：\n"
+            "- 用 search_demo_knowledge 检索，可换多组关键词多轮检索。\n"
+            "- 需要数值计算时用 calculate，不要心算。\n"
+            "- 检索不到就如实说「未检索到」，不要用常识补写。\n\n"
+            "输出格式：\n"
+            "- 先给一段不超过 5 句的结论。\n"
+            "- 再列来源编号与标题。\n"
+            "- 不要把检索到的原文整段复制回来，主线程不需要原始数据。"
+        ),
+        tools=[search_demo_knowledge, calculate, get_current_time],
+    )

extensions/registry.py

@@ -0,0 +1,199 @@
+"""确定性发现、声明和校验业务扩展；通过校验后一次性发布工具目录。"""
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+import importlib
+import re
+from pathlib import Path
+import threading
+from typing import Any, Callable, Iterable
+
+import config
+from tools import registry as tools
+
+_NAME = re.compile(r"[A-Za-z_][A-Za-z0-9_-]{0,63}\Z")
+_RESERVED_TOOLS = {"write_todos", "ls", "read_file", "write_file", "edit_file", "glob", "grep", "execute", "task"}
+_RESERVED_AGENTS = {"deep-agent", "general-purpose"}
+
+
+def _name(value: str, kind: str) -> str:
+    if not isinstance(value, str) or not _NAME.fullmatch(value):
+        raise ValueError(f"{kind}名称非法：{value!r}（使用字母、数字、下划线或连字符，最多 64 字符）")
+    return value
+
+
+def _text(value: str, kind: str) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"{kind}不能为空")
+    return value.strip()
+
+
+@dataclass(frozen=True)
+class AgentSpec:
+    name: str
+    description: str
+    prompt: str
+    tools: tuple[str, ...]
+    approval_tools: tuple[str, ...]
+
+
+class ExtensionRegistry:
+    """setup() 收到的注册器；不在导入业务模块时修改全局状态。"""
+
+    def __init__(self):
+        self.tools: dict[str, tools.ToolSpec] = {}
+        self.agents: dict[str, AgentSpec] = {}
+        self.modules: list[str] = []
+        self._sealed = False
+
+    def _editable(self):
+        if self._sealed:
+            raise ValueError("扩展已加载；新增或修改扩展后请重启服务")
+
+    def register_tool(self, func: Callable[..., Any], *, group: str | None = None,
+                      name: str | None = None, label: str | None = None,
+                      summary: str | None = None, requires_approval: bool | None = None) -> str:
+        """注册函数；参数覆盖 define_tool 元数据，也可以不使用装饰器。"""
+        self._editable()
+        spec = tools.tool_spec(func, group=group, name=name, label=label,
+                               summary=summary, requires_approval=requires_approval)
+        _name(spec.name, "工具")
+        if spec.name in _RESERVED_TOOLS:
+            raise ValueError(f"工具名与框架内置工具冲突：{spec.name}")
+        if spec.name in self.tools:
+            raise ValueError(f"工具名冲突：{spec.name}")
+        self.tools[spec.name] = spec
+        return spec.name
+
+    def _tool_name(self, value: str | Callable[..., Any]) -> str:
+        if isinstance(value, str):
+            return _name(value, "工具")
+        # 支持 register_agent(tools=[函数])；已显式注册的函数复用其注册名。
+        matches = [spec.name for spec in self.tools.values() if spec.func is value]
+        if len(matches) > 1:
+            raise ValueError("函数注册了多个名称，请使用明确的工具名引用")
+        return matches[0] if matches else self.register_tool(value)
+
+    def register_agent(self, *, name: str, description: str, prompt: str,
+                       tools: Iterable[str | Callable[..., Any]],
+                       approval_tools: Iterable[str | Callable[..., Any]] = ()) -> None:
+        """注册子 Agent；描述自动进入主 Agent 的委派说明，prompt 是子 Agent 的系统提示词。"""
+        self._editable()
+        _name(name, "Agent")
+        if name in _RESERVED_AGENTS or name in self.agents:
+            raise ValueError(f"Agent 名称冲突：{name}")
+        if isinstance(tools, str) or isinstance(approval_tools, str):
+            raise ValueError(f"Agent {name} 的 tools / approval_tools 必须是列表")
+        tool_names = tuple(self._tool_name(tool) for tool in tools)
+        approvals = []
+        for tool in approval_tools:
+            if isinstance(tool, str):
+                approvals.append(_name(tool, "工具"))
+            else:
+                matches = [spec.name for spec in self.tools.values() if spec.func is tool]
+                if len(matches) != 1:
+                    raise ValueError(f"Agent {name} 的审批工具未注册或名称不明确")
+                approvals.append(matches[0])
+        self.agents[name] = AgentSpec(name, _text(description, f"Agent {name} 描述"),
+                                      _text(prompt, f"Agent {name} 提示词"), tool_names, tuple(approvals))
+
+    def validate(self, extra_approvals: Iterable[str] = ()) -> set[str]:
+        """在所有模块声明完成后检查跨模块工具引用及审批，允许前向引用。"""
+        for spec in self.tools.values():
+            tools.spec_as_tool(spec)  # 提前验证函数 schema，错误不延迟到第一轮对话
+        approvals = {spec.name for spec in self.tools.values() if spec.requires_approval}
+        for agent in self.agents.values():
+            if len(set(agent.tools)) != len(agent.tools):
+                raise ValueError(f"Agent {agent.name} 重复引用工具")
+            missing = set(agent.tools) - self.tools.keys()
+            if missing:
+                raise ValueError(f"Agent {agent.name} 引用了未注册工具：{sorted(missing)}")
+            outside = set(agent.approval_tools) - set(agent.tools)
+            if outside:
+                raise ValueError(f"Agent {agent.name} 的审批工具未绑定：{sorted(outside)}")
+            approvals.update(agent.approval_tools)
+        approvals.update(extra_approvals)
+        missing = approvals - self.tools.keys()
+        if missing:
+            raise ValueError(f"审批配置引用了未注册工具：{sorted(missing)}")
+        return approvals
+
+    def subagents(self, approvals: set[str]) -> list[dict[str, Any]]:
+        return [{"name": agent.name, "description": agent.description,
+                 "system_prompt": agent.prompt,
+                 "tools": tools.get_tools_by_name(agent.tools),
+                 "interrupt_on": {name: True for name in agent.tools if name in approvals}}
+                for agent in self.agents.values()]
+
+
+_lock = threading.RLock()
+_loaded: ExtensionRegistry | None = None
+
+
+def discover_modules() -> list[str]:
+    """只扫描代码包下一级的 extension.py；不扫描用户上传或运行数据。"""
+    base = Path(__file__).resolve().parent
+    modules = []
+    for entry in sorted(base.iterdir()):
+        if entry.name.startswith(("_", ".")) or not entry.is_dir():
+            continue
+        path = entry / "extension.py"
+        if not path.exists():
+            continue
+        if entry.is_symlink() or path.is_symlink() or not entry.name.isidentifier():
+            raise ValueError(f"扩展目录非法或为符号链接：{entry.name}")
+        modules.append(f"extensions.{entry.name}.extension")
+    return modules
+
+
+def extra_approvals() -> list[str]:
+    return [name.strip() for name in config.get_str("HITL_TOOLS", "").split(",") if name.strip()]
+
+
+def load_extensions() -> ExtensionRegistry:
+    global _loaded
+    with _lock:
+        if _loaded is not None:
+            return _loaded
+        catalog = ExtensionRegistry()
+        for module_name in discover_modules():
+            try:
+                module = importlib.import_module(module_name)
+                setup = getattr(module, "setup", None)
+                if not callable(setup):
+                    raise ValueError("缺少 setup(registry) 函数")
+                setup(catalog)
+                catalog.modules.append(module_name)
+            except Exception as exc:
+                raise ValueError(f"扩展 {module_name} 加载失败：{exc}") from exc
+        approvals = catalog.validate(extra_approvals())
+        # 主提示词与恢复流程依赖这两个系统工具，缺失时必须在启动阶段报告。
+        missing = {"request_plan_approval", "ask_user"} - catalog.tools.keys()
+        if missing or not catalog.tools["request_plan_approval"].requires_approval:
+            raise ValueError("扩展必须注册 ask_user 和需要审批的 request_plan_approval")
+        # 工具清单展示的审批标记与实际执行规则保持一致。
+        catalog.tools = {name: replace(spec, requires_approval=name in approvals)
+                         for name, spec in catalog.tools.items()}
+        tools.install_specs(catalog.tools.values())
+        catalog._sealed = True
+        _loaded = catalog
+        return catalog
+
+
+def approval_names() -> set[str]:
+    catalog = load_extensions()
+    # 兼容通过旧 register() 入口增加的工具，并严格校验 HITL_TOOLS。
+    names = {spec.name for spec in catalog.tools.values() if spec.requires_approval}
+    names.update(name for agent in catalog.agents.values() for name in agent.approval_tools)
+    names.update(name for name in tools.get_tool_names() if tools.get_spec(name).requires_approval)
+    names.update(extra_approvals())
+    missing = names - set(tools.get_tool_names())
+    if missing:
+        raise ValueError(f"审批配置引用了未注册工具：{sorted(missing)}")
+    return names
+
+
+def reset_for_tests() -> None:
+    global _loaded
+    with _lock:
+        _loaded = None

main.py

@@ -2,7 +2,7 @@
 
 启动顺序（lifespan）：
 1. 日志
-2. 建表（幂等）
+2. 业务扩展发现与校验；建表（幂等）
 3. 计费单价表种子
 4. checkpointer 初始化
 5. **后台**预热 agent（图编译耗时百毫秒级，不能挡住 uvicorn 就绪）
@@ -89,6 +89,10 @@ async def _lifespan_startup() -> None:
         config.DB_DIALECT, config.MODEL, config.PORT,
     )
     _check_single_process()
+    # 扩展定义错误先阻断启动，避免启动调度器后才在后台预热时发现。
+    from extensions.registry import load_extensions, approval_names
+    load_extensions()
+    approval_names()
 
     # 建表与单价：失败即拒绝启动。半可用的服务比直接失败更难排查——
     # 用户会遇到「对话能开但一落库就报错」这类现象。

release/build.py

@@ -21,6 +21,7 @@ from common import sha256, validate_package, validation_env, version
 ROOT = Path(__file__).resolve().parents[1]
 ROOT_FILES = {"main.py", "config.py", "README.md", "docs/DEVELOPMENT.md", "requirements.txt", "pytest.ini", "VERSION", ".env.example"}
 TREE_SUFFIXES = {
+    "extensions": {".py", ".md", ".txt", ".json", ".yaml", ".yml"},
     "agents": {".py"}, "api": {".py"}, "billing": {".py"}, "db": {".py", ".sql"},
     "obs": {".py"}, "scheduler": {".py"}, "tools": {".py"},
     "skills": {".md", ".txt", ".json", ".yaml", ".yml"},

scripts/selfcheck.py

@@ -42,7 +42,7 @@ MODULES = [
     "config", "obs",
     "db.db_conn", "db.chat_store", "db.schedule_store", "db.billing_store",
     "billing.pricing", "billing.guard", "billing.meter",
-    "tools.registry", "tools.demo_tools",
+    "tools.registry", "tools.demo_tools", "extensions.registry",
     "agents.model_factory", "agents.prompts", "agents.middleware",
     "agents.agent_memory", "agents.agent_factory", "agents.agent_registry",
     "api.sse_events", "api.subagent_tracker", "api.context_meter",
@@ -120,7 +120,7 @@ def check_tools() -> str:
     count = registry.load_builtin_tools()
     if count == 0:
         raise RuntimeError(
-            "工具注册表为空——检查 tools/registry._BUILTIN_MODULES 与模块里的 @register"
+            "工具注册表为空——检查 extensions/*/extension.py 的 setup(registry)"
         )
     for tool in registry.get_all_tools():
         if not hasattr(tool, "name"):

tests/test_extensions.py

@@ -0,0 +1,182 @@
+"""业务扩展发现、注册、组装和启动失败路径。"""
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+import sys
+from types import SimpleNamespace
+
+import pytest
+
+from extensions import ExtensionRegistry, define_tool
+from extensions import registry as extensions
+from tools import registry as tools
+
+
+@define_tool(name="airflow", group="通风", label="计算风量")
+def airflow_impl(area: float, speed: float) -> float:
+    """计算断面积与风速对应的风量。"""
+    return area * speed
+
+
+def test_declaration_has_no_import_registration_and_alias_is_preserved():
+    assert "airflow" not in tools.get_tool_names()
+    catalog = ExtensionRegistry()
+    catalog.register_agent(name="ventilation", description="处理通风计算", prompt="使用工具计算风量。", tools=[airflow_impl])
+    catalog.validate()
+    tools.install_specs(catalog.tools.values())
+    tool = tools.get_all_tools()[0]
+    assert tool.name == "airflow"
+    assert tool.invoke({"area": 2, "speed": 3}) == 6
+    from agents.prompts import build_system_prompt
+    agents = catalog.subagents(set())
+    assert agents[0]["system_prompt"] == "使用工具计算风量。"
+    assert "处理通风计算" in build_system_prompt(agents)
+
+
+@pytest.mark.asyncio
+async def test_async_tool_alias():
+    @define_tool(name="async_airflow")
+    async def compute(area: float) -> float:
+        """异步计算。"""
+        return area * 2
+    catalog = ExtensionRegistry()
+    catalog.register_tool(compute)
+    catalog.validate()
+    tools.install_specs(catalog.tools.values())
+    tool = tools.get_all_tools()[0]
+    assert tool.name == "async_airflow"
+    assert await tool.ainvoke({"area": 3}) == 6
+
+
+def test_missing_reference_fails_and_forward_reference_works():
+    catalog = ExtensionRegistry()
+    catalog.register_agent(name="ventilation", description="通风", prompt="计算", tools=["airflow"])
+    with pytest.raises(ValueError, match="未注册工具"):
+        catalog.validate()
+    catalog.register_tool(airflow_impl)
+    catalog.validate()
+
+
+@pytest.mark.parametrize("name", ["bad name", "../escape", "1abc", "", "a" * 65])
+def test_invalid_names(name):
+    with pytest.raises(ValueError, match="名称非法"):
+        ExtensionRegistry().register_tool(airflow_impl, name=name)
+
+
+def test_duplicate_and_reserved_names():
+    catalog = ExtensionRegistry()
+    catalog.register_tool(airflow_impl)
+    with pytest.raises(ValueError, match="工具名冲突"):
+        catalog.register_tool(airflow_impl)
+    with pytest.raises(ValueError, match="内置工具冲突"):
+        catalog.register_tool(airflow_impl, name="write_file")
+    args = dict(name="ventilation", description="通风", prompt="计算", tools=["airflow"])
+    catalog.register_agent(**args)
+    with pytest.raises(ValueError, match="Agent 名称冲突"):
+        catalog.register_agent(**args)
+    with pytest.raises(ValueError, match="Agent 名称冲突"):
+        catalog.register_agent(**{**args, "name": "general-purpose"})
+
+
+@pytest.mark.parametrize("field", ["description", "prompt"])
+def test_empty_prompts_and_descriptions(field):
+    args = dict(name="ventilation", description="通风", prompt="计算", tools=[])
+    args[field] = " "
+    with pytest.raises(ValueError, match="不能为空"):
+        ExtensionRegistry().register_agent(**args)
+
+
+def test_approval_must_be_bound_and_configuration_must_exist():
+    catalog = ExtensionRegistry()
+    catalog.register_tool(airflow_impl)
+    catalog.register_agent(name="ventilation", description="通风", prompt="计算", tools=[], approval_tools=["airflow"])
+    with pytest.raises(ValueError, match="审批工具未绑定"):
+        catalog.validate()
+    with pytest.raises(ValueError, match="未注册工具"):
+        ExtensionRegistry().validate(["typo"])
+
+
+def test_shared_approval_applies_to_every_agent_using_tool():
+    catalog = ExtensionRegistry()
+    catalog.register_agent(name="one", description="一", prompt="一", tools=[airflow_impl], approval_tools=[airflow_impl])
+    catalog.register_agent(name="two", description="二", prompt="二", tools=[airflow_impl])
+    approvals = catalog.validate()
+    tools.install_specs(catalog.tools.values())
+    assert approvals == {"airflow"}
+    assert all(agent["interrupt_on"] == {"airflow": True} for agent in catalog.subagents(approvals))
+
+
+def test_failed_load_does_not_publish_partial_tools(monkeypatch):
+    baseline = tools.get_tool_names()
+    extensions.reset_for_tests()
+    def setup(catalog):
+        catalog.register_tool(airflow_impl)
+        raise ValueError("broken extension")
+    monkeypatch.setitem(sys.modules, "extensions.broken.extension", SimpleNamespace(setup=setup))
+    monkeypatch.setattr(extensions, "discover_modules", lambda: ["extensions.broken.extension"])
+    with pytest.raises(ValueError, match="extensions.broken.extension.*broken extension"):
+        extensions.load_extensions()
+    assert tools.get_tool_names() == baseline
+    assert extensions._loaded is None
+
+
+def test_missing_setup_fails_with_module_name(monkeypatch):
+    extensions.reset_for_tests()
+    monkeypatch.setitem(sys.modules, "extensions.broken.extension", SimpleNamespace())
+    monkeypatch.setattr(extensions, "discover_modules", lambda: ["extensions.broken.extension"])
+    with pytest.raises(ValueError, match="extensions.broken.extension.*setup"):
+        extensions.load_extensions()
+
+
+def test_multiple_cached_modules_reset_and_concurrent_load_are_idempotent(monkeypatch):
+    base = extensions.discover_modules()
+    calls = []
+    def setup(catalog):
+        calls.append(1)
+        catalog.register_tool(airflow_impl)
+    monkeypatch.setitem(sys.modules, "extensions.air.extension", SimpleNamespace(setup=setup))
+    monkeypatch.setattr(extensions, "discover_modules", lambda: [*base, "extensions.air.extension"])
+    for _ in range(2):
+        tools.reset_for_tests()
+        with ThreadPoolExecutor(max_workers=4) as pool:
+            results = list(pool.map(lambda _: extensions.load_extensions(), range(8)))
+        assert all(result is results[0] for result in results)
+        assert len(tools.get_tool_names()) == 6
+        assert "ask_user" in tools.get_tool_names() and "airflow" in tools.get_tool_names()
+    assert len(calls) == 2
+    with pytest.raises(ValueError, match="重启"):
+        results[0].register_tool(airflow_impl)
+
+
+def test_discovery_is_sorted_and_ignores_helpers(tmp_path, monkeypatch):
+    monkeypatch.setattr(extensions, "__file__", str(tmp_path / "registry.py"))
+    for name in ("zeta", "alpha", "_hidden"):
+        folder = tmp_path / name
+        folder.mkdir()
+        (folder / "extension.py").write_text("def setup(registry): pass")
+    (tmp_path / "helper.py").write_text("raise RuntimeError('must not import')")
+    assert extensions.discover_modules() == ["extensions.alpha.extension", "extensions.zeta.extension"]
+    (tmp_path / "linked").symlink_to(tmp_path / "alpha", target_is_directory=True)
+    with pytest.raises(ValueError, match="符号链接"):
+        extensions.discover_modules()
+
+
+@pytest.mark.asyncio
+async def test_invalid_extension_blocks_startup_before_database(monkeypatch):
+    import main
+    tools.reset_for_tests()
+    monkeypatch.setenv("HITL_TOOLS", "unknown_tool")
+    def forbidden():
+        pytest.fail("扩展校验失败后不应初始化数据库")
+    monkeypatch.setattr(main.db_conn, "init_schema", forbidden)
+    with pytest.raises(ValueError, match="未注册工具"):
+        await main._lifespan_startup()
+
+
+def test_release_includes_extension_code_and_prompt_files():
+    root = Path(__file__).resolve().parents[1]
+    sys.path.insert(0, str(root / "release"))
+    import build
+    files = {p.relative_to(root).as_posix() for p in build.source_files(root)}
+    assert "extensions/demo/extension.py" in files
+    assert "extensions/core/extension.py" in files
+    assert {".py", ".md", ".txt"} <= build.TREE_SUFFIXES["extensions"]

tests/test_runtime_regressions.py

@@ -490,3 +490,57 @@ async def test_cancellation_during_persistence_does_not_duplicate_answer(monkeyp
     assert len([m for m in chat_store.get_messages(session["id"]) if m["msg_role"] == "assistant"]) == 1
     assert len(billing_store.recent_usage("u")) == 1
     assert concurrency.stats()["global_active"] == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("delegated", [False, True])
+async def test_extension_approval_cannot_be_bypassed_by_main_or_subagent(monkeypatch, pricing_ready, delegated):
+    import sys
+    from extensions import registry as extensions
+    from extensions import define_tool
+    from tools import registry as tools
+    from agents import agent_factory, agent_memory
+    from main import app
+    writes = []
+
+    @define_tool(name="adjust_fan", group="通风")
+    def write_fan(speed: int) -> str:
+        """调整风机速度。"""
+        writes.append(speed)
+        return "完成"
+
+    def setup(catalog):
+        catalog.register_agent(name="ventilation", description="通风控制", prompt="使用工具调整风机。",
+                               tools=[write_fan], approval_tools=[write_fan])
+
+    modules = extensions.discover_modules()
+    monkeypatch.setitem(sys.modules, "extensions.ventilation.extension", SimpleNamespace(setup=setup))
+    monkeypatch.setattr(extensions, "discover_modules", lambda: [*modules, "extensions.ventilation.extension"])
+    tools.reset_for_tests()
+    responses = []
+    if delegated:
+        responses.append(AIMessage(content="", tool_calls=[{"id": "delegate", "name": "task",
+            "args": {"subagent_type": "ventilation", "description": "调整风机"}}]))
+    responses.extend([
+        AIMessage(content="", tool_calls=[{"id": "fan", "name": "adjust_fan", "args": {"speed": 10}}]),
+        AIMessage(content="已完成"), AIMessage(content="已汇总"),
+    ])
+    model = ScriptedModel(responses=responses)
+    monkeypatch.setattr("agents.model_factory.get_model", lambda: model)
+    graph = agent_factory.build_agent()
+    assert tools.get_spec("adjust_fan").requires_approval
+    monkeypatch.setattr("agents.agent_registry.get_agent", lambda: graph)
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+        first = await client.post("/api/chat", data={"message": "调整风机", "user_name": "u"})
+        events = [json.loads(line[6:]) for line in first.text.splitlines() if line.startswith("data: ")]
+        assert not [e for e in events if e["type"] == "error"], events
+        interrupted = next(e for e in events if e["type"] == "interrupt")
+        assert writes == []
+        sid = first.headers["x-session-id"]
+        token = interrupted["tool_approval"]["interrupt_token"]
+        response = await client.post("/api/chat/resume", json={"session_id": sid, "user_name": "u",
+            "interrupt_token": token, "approved": True})
+        assert response.status_code == 200, response.text
+        assert writes == [10], response.text
+        assert run_store.list_runs(sid)[0]["status"] == "success", response.text
+    await agent_memory.delete_thread(sid)

tools/__init__.py

@@ -1,7 +1,8 @@
 """工具层：注册表与内置工具。
 
 ``registry`` 是工具的唯一事实源；业务工具在 ``tools/`` 下新增模块并用
-``@register`` 标声明，再挂到 ``registry._BUILTIN_MODULES`` 里即可生效。
+用 ``define_tool`` 描述工具，再由 ``extensions/<业务>/extension.py`` 的
+``setup(registry)`` 统一注册。
 """
 
 from tools.registry import (

tools/demo_tools.py

@@ -25,14 +25,14 @@ import operator
 import re
 from typing import Any
 
-from tools.registry import register, source, sourced
+from tools.registry import define_tool, source, sourced
 
 # ============================================================
 # 1. 最简形态
 # ============================================================
 
 
-@register(group="系统", label="获取当前时间", summary="返回服务器当前本地时间")
+@define_tool(group="系统", label="获取当前时间", summary="返回服务器当前本地时间")
 def get_current_time() -> str:
     """返回服务器当前的本地日期时间。
 
@@ -74,7 +74,7 @@ _DEMO_KB: list[dict[str, Any]] = [
 ]
 
 
-@register(
+@define_tool(
     group="示例检索",
     label="检索演示知识库",
     summary="在内置演示知识库里按关键词检索，返回命中片段与来源",
@@ -172,7 +172,7 @@ def _eval_node(node: ast.AST) -> float:
     raise ValueError(f"不支持的表达式元素：{type(node).__name__}")
 
 
-@register(
+@define_tool(
     group="示例计算",
     label="安全算术计算",
     summary="对纯算术表达式做精确求值，避免心算出错",
@@ -215,16 +215,17 @@ def calculate(expression: str) -> dict[str, Any]:
 # 4. 人工审批（HITL 管道）
 # ============================================================
 
-# 这个工具的名字会被 agents/agent_factory 放进 interrupt_on：
+# requires_approval 声明会统一生成主 Agent 与子 Agent 的 interrupt_on：
 # 调用它时官方 HumanInTheLoopMiddleware 会在**执行前**拦下，向用户抛出
 # 审批请求；用户批准后才真正执行函数体。
 PLAN_APPROVAL_TOOL = "request_plan_approval"
 
 
-@register(
+@define_tool(
     group="人机协同",
     label="请求计划审批",
     summary="把待执行的计划提交给用户确认，用户批准后才继续",
+    requires_approval=True,
 )
 def request_plan_approval(plan: str) -> str:
     """把计划提交给用户审批，等待其批准后再继续执行。
@@ -245,7 +246,7 @@ def request_plan_approval(plan: str) -> str:
     return f"用户已批准以下计划，请按计划执行：\n{plan}"
 
 
-@register(group="人机协同", label="向用户提问", summary="缺少必要信息时暂停并向用户提问，收到回答后继续")
+@define_tool(group="人机协同", label="向用户提问", summary="缺少必要信息时暂停并向用户提问，收到回答后继续")
 def ask_user(questions: list[str]) -> dict[str, Any]:
     """当任务缺少必要信息且无法自行确定时，提出一至三个简短问题。
 

tools/registry.py

@@ -1,8 +1,7 @@
-"""工具的**唯一事实源**。
+"""工具元数据、运行目录与框架工具转换。
 
-全仓只有这里能声明「有哪些工具」。其他地方（agent 组装、系统提示词、
-工具名中文映射、SSE 事件渲染）都从这里派生，避免出现「提示词里说有、
-实际没注册」这类漂移。
+业务扩展通过 setup(registry) 统一声明，校验后发布到这里。
+Agent 组装、工具名中文映射和 SSE 事件渲染从同一目录读取。
 
 ## 来源标注契约
 
@@ -17,8 +16,8 @@ SSE 层（api/source_tracker.py）会从 ToolMessage 里抽取 ``sources`` 字
 
 ## 扩展方式
 
-新增业务工具：在 ``tools/`` 下新建模块，用 :func:`register` 装饰，
-然后 import 该模块（见本文件底部的 ``_load_builtin_modules``）。
+新增业务工具：使用 define_tool 描述元数据，并在 extensions/<业务>/extension.py
+的 setup(registry) 中声明工具与子 Agent。register 保留为兼容入口。
 """
 
 from __future__ import annotations
@@ -77,6 +76,56 @@ class _Registry:
 _registry = _Registry()
 
 
+def tool_spec(func: Callable[..., Any], *, group: str | None = None, name: str | None = None,
+              label: str | None = None, summary: str | None = None,
+              requires_approval: bool | None = None) -> ToolSpec:
+    """读取不带注册副作用的工具元数据；显式参数优先。"""
+    meta = getattr(func, "__vent_tool__", {})
+    selected = name if name is not None else meta.get("name") or getattr(func, "name", None) or getattr(func, "__name__", "")
+    approval = requires_approval if requires_approval is not None else meta.get("requires_approval", False)
+    if not isinstance(approval, bool):
+        raise ValueError(f"工具 {selected} 的 requires_approval 必须是布尔值")
+    return ToolSpec(name=selected, func=func,
+                    group=group if group is not None else meta.get("group", "业务工具"),
+                    label=label if label is not None else meta.get("label") or selected,
+                    summary=summary if summary is not None else meta.get("summary", ""),
+                    requires_approval=approval)
+
+
+def define_tool(*, group: str = "业务工具", label: str = "", name: str | None = None,
+                summary: str = "", requires_approval: bool = False):
+    """声明工具元数据，不在 import 时注册；函数仍可以直接调用和独立测试。"""
+    def decorator(func):
+        func.__vent_tool__ = dict(group=group, label=label, name=name, summary=summary,
+                                  requires_approval=requires_approval)
+        return func
+    return decorator
+
+
+def spec_as_tool(spec: ToolSpec) -> Any:
+    """保留注册别名、完整 docstring 和异步执行方式，转换成框架工具。"""
+    from langchain_core.tools import BaseTool, StructuredTool
+    if isinstance(spec.func, BaseTool):
+        return spec.func.model_copy(update={"name": spec.name})
+    description = inspect.getdoc(spec.func) or spec.summary
+    if not description:
+        raise ValueError(f"工具 {spec.name} 缺少 docstring 或 summary")
+    is_async = inspect.iscoroutinefunction(spec.func)
+    return StructuredTool.from_function(
+        func=None if is_async else spec.func, coroutine=spec.func if is_async else None,
+        name=spec.name, description=description,
+    )
+
+
+def install_specs(specs: Iterable[ToolSpec]) -> None:
+    """校验完成后一次性替换全局目录，加载失败不发布半个注册表。"""
+    global _registry
+    replacement = _Registry()
+    for spec in specs:
+        replacement.add(spec)
+    _registry = replacement
+
+
 # ============================================================
 # 注册
 # ============================================================
@@ -160,7 +209,7 @@ def as_tool(func: Callable[..., Any]) -> Any:
 
 def get_all_tools() -> list[Any]:
     """全部工具，以 ``BaseTool`` 形态返回。主 agent 用这个。"""
-    return [as_tool(spec.func) for spec in _registry.tools.values()]
+    return [spec_as_tool(spec) for spec in _registry.tools.values()]
 
 
 def get_tools_by_name(names: Iterable[str]) -> list[Any]:
@@ -175,7 +224,7 @@ def get_tools_by_name(names: Iterable[str]) -> list[Any]:
         if spec is None:
             logger.warning("子智能体引用了未注册的工具 %r，已跳过", name)
             continue
-        tools.append(as_tool(spec.func))
+        tools.append(spec_as_tool(spec))
     return tools
 
 
@@ -258,33 +307,16 @@ def sourced(
 # 内置工具模块装载
 # ============================================================
 
-_BUILTIN_MODULES = ("tools.demo_tools",)
-
-
 def load_builtin_tools() -> int:
-    """import 内置工具模块以触发注册。返回注册到的工具数。
-
-    这里要处理一个 Python 导入语义的坑：模块的注册是靠**模块体里的装饰器**
-    完成的，而 ``import_module`` 对已导入的模块是空操作——模块一旦在
-    ``sys.modules`` 里，再 import 不会重放装饰器。所以「注册表被清空后重新
-    装载」这个场景（测试隔离、或将来做热重载）必须显式 ``reload``，
-    否则注册表会一直是空的，表现为「工具莫名其妙全没了」。
-    """
-    import importlib
-    import sys
-
-    for module_name in _BUILTIN_MODULES:
-        if module_name not in sys.modules:
-            importlib.import_module(module_name)
-        elif not _registry.tools:
-            # 模块已在缓存里但注册表是空的 → 重放模块体以重新注册。
-            # 生产路径不会走到这里（首次 import 就已填充注册表），
-            # 只在测试 reset 之后触发。
-            importlib.reload(sys.modules[module_name])
+    """兼容旧调用方；工具和子 Agent 统一由业务扩展声明。"""
+    from extensions.registry import load_extensions
+    load_extensions()
     return len(_registry.tools)
 
 
 def reset_for_tests() -> None:
-    """仅供测试：清空注册表，使下一次 ``load_builtin_tools`` 重新注册。"""
+    """仅供测试：清空发布目录；下次重新执行 setup，不依赖 import 重放。"""
+    from extensions.registry import reset_for_tests as reset_extensions
     _registry.tools.clear()
     _registry.groups.clear()
+    reset_extensions()