feat: 支持扩展服务生命周期与中间件注册

.env.example

@@ -97,3 +97,6 @@ AUTH_ENABLED=0
 # 观测
 # ============================================================
 TFA_TRACE_MIDDLEWARE=1
+
+# 离线扩展示例：共享术语表服务 + 工具 + 上下文中间件，默认关闭
+CONTEXT_DEMO_ENABLED=0

README.md

@@ -134,7 +134,7 @@ DEEP_AGENT_ENV_PRIORITY=process LOG_LEVEL=DEBUG ~/vent-agent/current/release/ser
 
 聊天、定时任务和批处理共用 `runtime/` 执行层；HTTP 层负责 SSE 编码。独立脚本使用 `RuntimeEnvironment` 自动初始化和清理资源，默认不启动调度器。调用方式见 [执行层与入口](docs/DEVELOPMENT.md#执行层与入口)。
 
-新增业务时，在 `extensions/<业务名>/extension.py` 编写 `setup(registry)`，统一声明工具、子 Agent、提示词和审批要求。启动自动发现并校验，无需修改工厂函数；修改后重启生效。示例和完整步骤见 [扩展方式](docs/DEVELOPMENT.md#扩展方式)。
+新增业务时，在 `extensions/<业务名>/extension.py` 编写 `setup(registry)`，统一声明工具、子 Agent、提示词、审批、共享服务和中间件。启动自动发现并校验，无需修改工厂函数；修改后重启生效。RAG/记忆接入可参考默认关闭的 `extensions/context_demo/` 离线示例。完整步骤见 [扩展方式](docs/DEVELOPMENT.md#扩展方式)。
 
 ```bash
 .venv/bin/python -m pytest            # 离线测试，使用临时数据库

agents/agent_factory.py

@@ -21,6 +21,7 @@ from agents.middleware import DynamicTimeMiddleware
 from obs import get_logger
 from tools import registry as tool_registry
 from extensions.registry import approval_names, load_extensions
+from extensions.services import service_view
 
 logger = get_logger(__name__)
 
@@ -120,11 +121,11 @@ def _apply_compaction_setting() -> None:
 def _build_middleware() -> list[Any]:
     """用户中间件栈。
 
-    只放框架没有的能力（动态时间）。压缩交给框架内置的摘要中间件处理，
+    动态时间后接入扩展声明的中间件。默认压缩交给框架内置的摘要中间件处理，
     见 :func:`_apply_compaction_setting`。
     """
     _apply_compaction_setting()
-    return [DynamicTimeMiddleware()]
+    return [DynamicTimeMiddleware(), *load_extensions().build_middleware("main")]
 
 
 def _build_interrupt_on() -> dict[str, Any]:
@@ -144,6 +145,10 @@ def build_agent() -> Any:
     tool_registry.load_builtin_tools()
     tools = tool_registry.get_all_tools()
     subagents = _build_subagents()
+    catalog = load_extensions()
+    service_view(catalog.services)  # 启用了外部服务时，必须先进入运行环境。
+    for spec in subagents:
+        spec["middleware"] = catalog.build_middleware(spec["name"])
 
     agent = create_deep_agent(
         model=model_factory.get_model(),
@@ -175,6 +180,10 @@ def describe() -> dict[str, Any]:
     return {
         "name": "deep-agent",
         "extensions": list(load_extensions().modules),
+        "disabled_extensions": list(load_extensions().disabled_modules),
+        "services": sorted(load_extensions().services),
+        "middleware": [{"name": name, "targets": list(load_extensions().middleware[name].targets)}
+                       for name in load_extensions().middleware_order()],
         "model": model_factory.describe(),
         "tools": tool_registry.describe(),
         "interrupt_on": sorted(_build_interrupt_on().keys()),

docs/DEVELOPMENT.md

@@ -213,7 +213,7 @@ if __name__ == "__main__":
 
 ### 公共运行环境
 
-`RuntimeEnvironment` 负责加载和校验扩展、初始化业务库与计费单价、标记上次意外中断的执行、打开 checkpoint，并在后台预热 Agent、重试待结算流水。退出时停止可选调度器，等待维护和编译完成，清除 Agent 缓存，关闭 checkpoint 与业务库连接。启动失败或调用方取消时同样清理。
+`RuntimeEnvironment` 负责加载和校验扩展、初始化业务库与计费单价、标记上次意外中断的执行、打开 checkpoint，并在后台预热 Agent、重试待结算流水。退出时停止可选调度器，等待维护和编译完成，清除 Agent 缓存，关闭扩展服务、checkpoint 与业务库连接。启动失败或调用方取消时同样清理。
 
 | 参数 | 默认值 | 用途 |
 |---|---|---|
@@ -275,6 +275,76 @@ def setup(registry):
 
 内置示例见 `extensions/demo/extension.py`；`extensions/core/extension.py` 注册计划审批和询问用户工具，主提示词与恢复流程依赖它们，需保留。
 
+### 扩展服务与中间件
+
+RAG 客户端、记忆存储等共享资源通过 `register_service()` 声明，由 `RuntimeEnvironment` 启动和关闭。模型调用前后处理通过 `register_middleware()` 接入，无需修改主入口和 Agent 工厂。
+
+完整离线示例见 `extensions/context_demo/extension.py`：共享术语表 → 检索工具 → 模型调用前补充上下文。在 `.env` 添加 `CONTEXT_DEMO_ENABLED=1` 后重启即可体验；默认关闭，不访问外部服务。这只是接口示例，真实知识库的索引、长期记忆存储和上下文预算仍需业务模块实现。
+
+扩展可提供三个同步入口：
+
+```python
+def enabled():
+    return config.get_bool("MY_RAG_ENABLED", False)
+
+
+def validate_config():
+    # 仅检查配置，不开连接；抛 ValueError 会阻止启动。
+    if not config.get_str("MY_RAG_URL", ""):
+        raise ValueError("启用 RAG 时必须配置 MY_RAG_URL")
+
+
+def setup(registry):
+    registry.register_service("rag_client", open_rag_client)
+    registry.register_tool(search_knowledge, requires=["rag_client"])
+    registry.register_middleware(
+        "RecallMiddleware", make_recall_middleware,
+        requires=["rag_client"], targets=["main", "researcher"],
+    )
+```
+
+以上是接口示意；`researcher` 必须由当前或其他启用的扩展注册。`enabled()` 和 `validate_config()` 都可省略；省略开关默认启用。关闭模块会跳过配置校验与 `setup()`，因此也不会注册其工具、服务或中间件。模块本身仍需导入才能读取开关，可选第三方依赖应放到服务工厂或中间件工厂内延迟导入。
+
+**服务工厂**采用异步上下文管理器：
+
+```python
+from contextlib import asynccontextmanager
+from extensions import get_service
+
+
+@asynccontextmanager
+async def open_rag_client(dependencies):
+    # Client 为所选知识库 SDK 的客户端，在这里延迟导入。
+    client = Client(config.get_str("MY_RAG_URL"))
+    try:
+        yield client
+    finally:
+        await client.aclose()
+
+
+async def search_knowledge(query: str) -> dict:
+    """检索业务知识库，返回片段与来源。"""
+    client = get_service("rag_client")
+    return await client.search(query)
+```
+
+- 工厂参数是只读映射，仅包含 `requires` 声明的服务。服务可以依赖其他服务，按依赖顺序启动，逆序关闭。
+- 工具在调用时通过 `get_service()` 获取资源；不要在导入或 `setup()` 阶段取服务。有服务依赖的工具先显式注册，再用名称或函数绑定到子 Agent。
+- 服务开始初始化后，即使调用方取消，也会等待初始化收尾再清理。工厂在 `yield` 前失败时，必须在自己的 `try/finally` 中释放已申请的资源。
+- 服务是进程内共享实例。异步客户端应由异步工具在同一事件循环中调用；用户、会话相关数据需要明确作用域，不能直接存进共享实例或中间件字段。执行身份可从工具的 `ToolRuntime.config["configurable"]` 中读取 `workspace_user`、`workspace_session`。
+- 环境退出会等待调度、预热和计费任务，清除 Agent 缓存，然后关闭扩展服务、checkpoint 和业务库；独立脚本创建的任务仍需自行等待完成。
+
+**中间件工厂**使用同步函数 `factory(services)`，返回一个 `AgentMiddleware`，实例的 `name` 必须与注册名一致。
+
+- `targets` 默认 `["main"]`；可填写已注册子 Agent 名称。没有隐式继承，不支持向框架自动创建的 `general-purpose` 注入。
+- `requires` 声明需要使用的服务。模型或工具包装方法应实现对应的异步钩子，框架实际通过异步路径执行。
+- `before=["OtherMiddleware"]`、`after=["OtherMiddleware"]` 控制扩展中间件列表顺序；引用必须存在并有共同目标。没有顺序约束时按名称稳定排序。
+- 对 `wrap_model_call`，列表靠前者在外层：进入时按顺序，返回时逆序。它们仍位于框架内置中间件之后；主 Agent 的动态时间中间件位于扩展之前。
+- 启动时会试构造一次，Agent 编译时为每个目标重新构造。因此工厂必须无副作用，每次返回新实例；连接、后台任务应由服务管理。Agent 本身为单例，中间件不能在实例字段中保存某次请求的状态。
+- 框架内置中间件名称被保留，重复摘要/权限/审批中间件会被拒绝。默认压缩行为保持原样。后续自定义压缩应使用独立名称并显式关闭内置压缩（`CTX_COMPACT_ENABLED=0`），同时自行保留历史归档与工具调用配对等约束；当前没有通用的内置中间件替换接口。
+
+`GET /api/model` 的 `agent` 对象会列出 `services`、有序的 `middleware` 及其目标，以及 `disabled_extensions`；不返回服务实例和连接配置。修改扩展或启用开关后重启服务。
+
 ### 声明审批
 
 有写入或外部副作用的工具使用 `@define_tool(..., requires_approval=True)`。也可以在 Agent 定义中设置 `approval_tools=[工具函数或名称]`，但必须包含在该 Agent 的 `tools` 中。
@@ -288,7 +358,10 @@ def setup(registry):
 - 工具和 Agent 名称格式、重名、与框架内置名称冲突。
 - 工具函数能否生成参数 schema，是否提供 docstring 或 summary。
 - Agent 描述和提示词非空，工具引用有效，审批工具确实绑定。
-- `setup()` 缺失或加载失败、目录/入口符号链接等问题。
+- 服务依赖、中间件目标及顺序引用有效，没有循环、重名或内置名称冲突。
+- 启用模块的配置校验、`setup()` 缺失或加载失败、目录/入口符号链接等问题。
+
+数据库与 checkpoint 初始化后，按依赖启动扩展服务，再验证中间件工厂的实际返回类型与名称；失败会释放已启动资源，不进入调度与预热阶段。
 
 新增或修改扩展后**重启服务**，再用 `GET /api/model` 查看 `agent.extensions` 和子 Agent 列表，用 `GET /api/model/tools` 查看工具。`/api/model/reload` 只重建已加载定义，不重新导入业务代码。
 

extensions/__init__.py

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

extensions/context_demo/__init__.py

@@ -0,0 +1 @@
+"""默认关闭的服务与中间件接入示例。"""

extensions/context_demo/extension.py

@@ -0,0 +1,57 @@
+"""离线示例：共享术语表服务、检索工具和模型调用前的上下文补充。"""
+from contextlib import asynccontextmanager
+
+import config
+from extensions import define_tool, get_service
+
+
+def enabled() -> bool:
+    return config.get_bool("CONTEXT_DEMO_ENABLED", False)
+
+
+@asynccontextmanager
+async def glossary_service(dependencies):
+    # 实际接入时在这里延迟导入客户端、打开连接，并在 finally 中关闭。
+    glossary = {"风量": "单位时间内通过某一断面的空气体积。", "风速": "空气流动的速度。"}
+    try:
+        yield glossary
+    finally:
+        glossary.clear()
+
+
+@define_tool(group="扩展示例", label="查询术语")
+def lookup_demo_term(term: str) -> str:
+    """查询离线演示术语表，仅用于演示模块接入。"""
+    return get_service("demo_glossary").get(term, "术语表中没有该词条")
+
+
+def glossary_middleware(services):
+    # 模块关闭时不会加载可选依赖。工厂只组装中间件，不创建连接。
+    from langchain.agents.middleware.types import AgentMiddleware
+    class DemoGlossaryMiddleware(AgentMiddleware):
+        def enrich(self, request):
+            message = request.system_message
+            if message is None:
+                return request
+            terms = "、".join(sorted(services["demo_glossary"]))
+            text = f"\n演示术语表可查询：{terms}。需要时调用 lookup_demo_term。"
+            content = (message.content + text if isinstance(message.content, str)
+                       else [*message.content, {"type": "text", "text": text}])
+            return request.override(system_message=message.model_copy(update={
+                "content": content,
+            }))
+
+        def wrap_model_call(self, request, handler):
+            return handler(self.enrich(request))
+
+        async def awrap_model_call(self, request, handler):
+            return await handler(self.enrich(request))
+
+    return DemoGlossaryMiddleware()
+
+
+def setup(registry):
+    registry.register_service("demo_glossary", glossary_service)
+    registry.register_tool(lookup_demo_term, requires=["demo_glossary"])
+    registry.register_middleware("DemoGlossaryMiddleware", glossary_middleware,
+                                requires=["demo_glossary"], targets=["main"])

extensions/registry.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass, replace
 import importlib
+import inspect
 import re
 from pathlib import Path
 import threading
@@ -10,10 +11,36 @@ from typing import Any, Callable, Iterable
 
 import config
 from tools import registry as tools
+from extensions.services import ServiceSpec, ordered_names, service_view
 
 _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"}
+_RESERVED_AGENTS = {"deep-agent", "general-purpose", "main"}
+_RESERVED_MIDDLEWARE = {
+    "DynamicTimeMiddleware", "SummarizationMiddleware", "TodoListMiddleware",
+    "FilesystemMiddleware", "SubAgentMiddleware", "PatchToolCallsMiddleware",
+    "SkillsMiddleware", "MemoryMiddleware", "HumanInTheLoopMiddleware",
+    "AnthropicPromptCachingMiddleware",
+}
+
+
+def _names(values: Iterable[str], kind: str) -> tuple[str, ...]:
+    if isinstance(values, str):
+        raise ValueError(f"{kind}必须是名称列表")
+    result = tuple(_name(value, kind) for value in values)
+    if len(set(result)) != len(result):
+        raise ValueError(f"{kind}不能重复")
+    return result
+
+
+@dataclass(frozen=True)
+class MiddlewareSpec:
+    name: str
+    factory: Callable[..., Any]
+    targets: tuple[str, ...]
+    requires: tuple[str, ...]
+    before: tuple[str, ...]
+    after: tuple[str, ...]
 
 
 def _name(value: str, kind: str) -> str:
@@ -28,6 +55,15 @@ def _text(value: str, kind: str) -> str:
     return value.strip()
 
 
+def _call_sync(hook: Callable[..., Any], name: str, *args) -> Any:
+    result = hook(*args)
+    if inspect.isawaitable(result):
+        if inspect.iscoroutine(result):
+            result.close()
+        raise ValueError(f"{name} 必须同步执行；异步初始化请放入服务工厂")
+    return result
+
+
 @dataclass(frozen=True)
 class AgentSpec:
     name: str
@@ -44,6 +80,10 @@ class ExtensionRegistry:
         self.tools: dict[str, tools.ToolSpec] = {}
         self.agents: dict[str, AgentSpec] = {}
         self.modules: list[str] = []
+        self.disabled_modules: list[str] = []
+        self.services: dict[str, ServiceSpec] = {}
+        self.middleware: dict[str, MiddlewareSpec] = {}
+        self.tool_services: dict[str, tuple[str, ...]] = {}
         self._sealed = False
 
     def _editable(self):
@@ -52,7 +92,8 @@ class ExtensionRegistry:
 
     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:
+                      summary: str | None = None, requires_approval: bool | None = None,
+                      requires: Iterable[str] = ()) -> str:
         """注册函数；参数覆盖 define_tool 元数据，也可以不使用装饰器。"""
         self._editable()
         spec = tools.tool_spec(func, group=group, name=name, label=label,
@@ -62,9 +103,85 @@ class ExtensionRegistry:
             raise ValueError(f"工具名与框架内置工具冲突：{spec.name}")
         if spec.name in self.tools:
             raise ValueError(f"工具名冲突：{spec.name}")
+        self.tool_services[spec.name] = _names(requires, "工具服务依赖")
         self.tools[spec.name] = spec
         return spec.name
 
+    def register_service(self, name: str, factory: Callable[..., Any], *,
+                         requires: Iterable[str] = ()) -> None:
+        """声明服务。factory(dependencies) 返回 async context manager。"""
+        self._editable()
+        _name(name, "服务")
+        if name in self.services:
+            raise ValueError(f"服务名称冲突：{name}")
+        if not callable(factory) or inspect.iscoroutinefunction(factory):
+            raise ValueError(f"服务 {name} 的 factory 必须返回异步上下文管理器，请使用 asynccontextmanager")
+        self.services[name] = ServiceSpec(name, factory, _names(requires, "服务依赖"))
+
+    def register_middleware(self, name: str, factory: Callable[..., Any], *,
+                            targets: Iterable[str] = ("main",), requires: Iterable[str] = (),
+                            before: Iterable[str] = (), after: Iterable[str] = ()) -> None:
+        """声明无副作用的同步工厂；每次编译为每个目标生成独立中间件。"""
+        self._editable()
+        _name(name, "中间件")
+        if name in self.middleware or name in _RESERVED_MIDDLEWARE:
+            raise ValueError(f"中间件名称冲突：{name}")
+        if not callable(factory) or inspect.iscoroutinefunction(factory):
+            raise ValueError(f"中间件 {name} 的 factory 必须是同步函数")
+        targets = _names(targets, "中间件目标")
+        if not targets:
+            raise ValueError(f"中间件 {name} 至少需要一个目标")
+        self.middleware[name] = MiddlewareSpec(name, factory, targets,
+            _names(requires, "中间件服务依赖"), _names(before, "before"), _names(after, "after"))
+
+    def middleware_order(self) -> list[str]:
+        deps = {name: set(spec.after) for name, spec in self.middleware.items()}
+        for name, spec in self.middleware.items():
+            for ref in (*spec.before, *spec.after):
+                if ref not in deps:
+                    raise ValueError(f"中间件 {name} 引用未注册项：{ref}")
+                if not set(spec.targets).intersection(self.middleware[ref].targets):
+                    raise ValueError(f"中间件 {name} 与 {ref} 没有共同目标，不能指定顺序")
+            for ref in spec.before:
+                deps[ref].add(name)
+        return ordered_names(deps, "中间件")
+
+    def build_middleware(self, target: str) -> list[Any]:
+        from langchain.agents.middleware.types import AgentMiddleware
+        result = []
+        for name in self.middleware_order():
+            spec = self.middleware[name]
+            if target not in spec.targets:
+                continue
+            instance = spec.factory(service_view(spec.requires))
+            if not isinstance(instance, AgentMiddleware) or instance.name != name:
+                if inspect.iscoroutine(instance):
+                    instance.close()
+                raise ValueError(f"中间件 {name} 的工厂必须返回同名 AgentMiddleware 实例")
+            result.append(instance)
+        return result
+
+    def validate_components(self) -> None:
+        ordered_names({name: set(spec.requires) for name, spec in self.services.items()}, "服务")
+        for name, requires in self.tool_services.items():
+            self._check_services(f"工具 {name}", requires)
+        for name, spec in self.middleware.items():
+            self._check_services(f"中间件 {name}", spec.requires)
+            missing = set(spec.targets) - {"main", *self.agents}
+            if missing:
+                raise ValueError(f"中间件 {name} 引用了未注册目标：{sorted(missing)}")
+        self.middleware_order()
+
+    def _check_services(self, owner: str, names: Iterable[str]) -> None:
+        missing = set(names) - self.services.keys()
+        if missing:
+            raise ValueError(f"{owner} 引用了未注册服务：{sorted(missing)}")
+
+    def check_middleware_factories(self) -> None:
+        # 在服务就绪后验证工厂；编译时重新创建，不能在工厂里开连接或启动任务。
+        for target in ("main", *sorted(self.agents)):
+            self.build_middleware(target)
+
     def _tool_name(self, value: str | Callable[..., Any]) -> str:
         if isinstance(value, str):
             return _name(value, "工具")
@@ -99,6 +216,7 @@ class ExtensionRegistry:
 
     def validate(self, extra_approvals: Iterable[str] = ()) -> set[str]:
         """在所有模块声明完成后检查跨模块工具引用及审批，允许前向引用。"""
+        self.validate_components()
         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}
@@ -159,10 +277,25 @@ def load_extensions() -> ExtensionRegistry:
         for module_name in discover_modules():
             try:
                 module = importlib.import_module(module_name)
+                enabled = getattr(module, "enabled", None)
+                if enabled is not None:
+                    if not callable(enabled):
+                        raise ValueError("enabled 必须是返回布尔值的函数")
+                    active = _call_sync(enabled, "enabled")
+                    if type(active) is not bool:
+                        raise ValueError("enabled() 必须返回布尔值")
+                    if not active:
+                        catalog.disabled_modules.append(module_name)
+                        continue
                 setup = getattr(module, "setup", None)
                 if not callable(setup):
                     raise ValueError("缺少 setup(registry) 函数")
-                setup(catalog)
+                validate_config = getattr(module, "validate_config", None)
+                if validate_config is not None:
+                    if not callable(validate_config) or inspect.iscoroutinefunction(validate_config):
+                        raise ValueError("validate_config 必须是同步校验函数")
+                    _call_sync(validate_config, "validate_config")
+                _call_sync(setup, "setup", catalog)
                 catalog.modules.append(module_name)
             except Exception as exc:
                 raise ValueError(f"扩展 {module_name} 加载失败：{exc}") from exc

extensions/services.py

@@ -0,0 +1,94 @@
+"""扩展服务的依赖顺序、共享实例和异步资源管理。"""
+from __future__ import annotations
+
+from contextlib import AsyncExitStack
+from dataclasses import dataclass
+import inspect
+from types import MappingProxyType
+from typing import Any, AsyncContextManager, Callable, Mapping
+
+
+@dataclass(frozen=True)
+class ServiceSpec:
+    name: str
+    factory: Callable[[Mapping[str, Any]], AsyncContextManager[Any]]
+    requires: tuple[str, ...] = ()
+
+
+def ordered_names(dependencies: Mapping[str, set[str]], kind: str) -> list[str]:
+    """稳定拓扑排序：无依赖的节点按名称排序，缺失引用和循环立即报错。"""
+    pending = {name: set(refs) for name, refs in dependencies.items()}
+    for name, refs in pending.items():
+        missing = refs - pending.keys()
+        if missing:
+            raise ValueError(f"{kind} {name} 引用未注册项：{sorted(missing)}")
+    result = []
+    while pending:
+        ready = sorted(name for name, refs in pending.items() if not refs)
+        if not ready:
+            raise ValueError(f"{kind}存在循环依赖：{sorted(pending)}")
+        for name in ready:
+            result.append(name)
+            del pending[name]
+        for refs in pending.values():
+            refs.difference_update(ready)
+    return result
+
+
+_active: ServiceContainer | None = None
+
+
+class ServiceContainer:
+    """由 RuntimeEnvironment 管理；工厂必须返回异步上下文管理器。
+
+    工厂只收到其声明的依赖。资源先按依赖顺序创建，再逆序关闭。
+    """
+
+    def __init__(self, specs: Mapping[str, ServiceSpec]):
+        self._specs = dict(specs)
+        self._values: dict[str, Any] = {}
+        self._stack = AsyncExitStack()
+
+    async def start(self) -> None:
+        global _active
+        if _active is not None:
+            raise RuntimeError("扩展服务已启动，请复用 RuntimeEnvironment")
+        order = ordered_names({name: set(spec.requires) for name, spec in self._specs.items()}, "服务")
+        for name in order:
+            spec = self._specs[name]
+            dependencies = self.view(spec.requires)
+            context = spec.factory(dependencies)
+            if not hasattr(context, "__aenter__") or not hasattr(context, "__aexit__"):
+                if inspect.iscoroutine(context):
+                    context.close()
+                raise TypeError(f"服务 {name} 的工厂必须返回异步上下文管理器")
+            self._values[name] = await self._stack.enter_async_context(context)
+        _active = self
+
+    def view(self, names) -> Mapping[str, Any]:
+        try:
+            return MappingProxyType({name: self._values[name] for name in names})
+        except KeyError as exc:
+            raise RuntimeError(f"扩展服务 {exc.args[0]} 尚未就绪，请在 RuntimeEnvironment 内使用") from exc
+
+    async def close(self) -> None:
+        global _active
+        try:
+            await self._stack.aclose()
+        finally:
+            self._values.clear()
+            if _active is self:
+                _active = None
+
+
+def service_view(names) -> Mapping[str, Any]:
+    if not names:
+        return MappingProxyType({})
+    if _active is None:
+        raise RuntimeError("扩展服务尚未启动，请在 RuntimeEnvironment 内使用")
+    return _active.view(names)
+
+
+def get_service(name: str) -> Any:
+    """工具在执行时获取服务，不要在 import 或 setup 时调用。"""
+    return service_view((name,))[name]

runtime/environment.py

@@ -11,6 +11,7 @@ from agents import agent_memory, agent_registry
 from billing import pricing
 from db import db_conn, run_store
 from extensions.registry import approval_names, load_extensions
+from extensions.services import ServiceContainer
 from obs import get_logger, setup_logging
 
 logger = get_logger(__name__)
@@ -86,7 +87,7 @@ class RuntimeEnvironment:
             setup_logging(config.LOG_LEVEL, config.LOG_FORMAT)
             config.log_config_sources()
             _check_single_process()
-            load_extensions()
+            catalog = load_extensions()
             approval_names()
 
             stack.callback(db_conn.close)
@@ -99,12 +100,19 @@ class RuntimeEnvironment:
             # 注册在初始化前，部分初始化失败也会清理。
             stack.push_async_callback(agent_memory.close_checkpointer)
             agent_registry.invalidate()
-            stack.callback(agent_registry.invalidate)
             init_task = asyncio.create_task(agent_memory.init_checkpointer(), name="runtime-checkpoint-init")
             stack.push_async_callback(_join, init_task)
             # aiosqlite 在线程中连接；取消调用方时先等初始化结束，再关闭连接。
             await asyncio.shield(init_task)
 
+            services = ServiceContainer(catalog.services)
+            stack.push_async_callback(services.close)
+            task = asyncio.create_task(services.start(), name="runtime-services-init")
+            stack.push_async_callback(_join, task)
+            await asyncio.shield(task)
+            stack.callback(agent_registry.invalidate)
+            catalog.check_middleware_factories()
+
             if self.prewarm:
                 task = asyncio.create_task(_prewarm(), name="runtime-prewarm")
                 stack.push_async_callback(_join, task)

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", "extensions.registry",
+    "tools.registry", "tools.demo_tools", "extensions.registry", "extensions.services",
     "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",

tests/conftest.py

@@ -26,6 +26,7 @@ os.environ.setdefault("DB_DIALECT", "sqlite")
 os.environ.setdefault("DB_SQLITE_PATH", str(_TMP_ROOT / "test.sqlite"))
 os.environ.setdefault("CHECKPOINTER", "memory")   # 测试不需要落盘图状态
 os.environ["CTX_COMPACT_ENABLED"] = "0"           # 不引入摘要中间件，测试更纯粹
+os.environ["CONTEXT_DEMO_ENABLED"] = "0"          # 示例扩展由专门用例启用
 os.environ["OPENAI_API_KEY"] = "sk-test-key"
 os.environ["OPENAI_BASE_URL"] = "https://example.invalid"
 os.environ["TFAGENTS_DAILY_TOKEN_LIMIT"] = "0"

tests/test_environment.py

@@ -27,8 +27,10 @@ def lifecycle(monkeypatch):
         async def call(*args, **kwargs):
             calls.append(name)
         return call
-    for name in ('setup_logging', 'load_extensions', 'approval_names'):
+    for name in ('setup_logging', 'approval_names'):
         monkeypatch.setattr(environment, name, mark(name))
+    from extensions.registry import ExtensionRegistry
+    monkeypatch.setattr(environment, 'load_extensions', lambda: ExtensionRegistry())
     monkeypatch.setattr(config, 'log_config_sources', mark('config'))
     monkeypatch.setattr(db_conn, 'init_schema', mark('schema'))
     monkeypatch.setattr(db_conn, 'health', lambda: {'status': 'ok'})
@@ -128,7 +130,7 @@ async def test_cancel_during_startup_waits_for_initialization(lifecycle, monkeyp
     finish.set()
     with pytest.raises(asyncio.CancelledError):
         await task
-    assert lifecycle[-4:] == ['checkpoint', 'invalidate', 'checkpoint_close', 'db_close']
+    assert lifecycle[-3:] == ['checkpoint', 'checkpoint_close', 'db_close']
 
 
 @pytest.mark.asyncio

tests/test_extension_components.py

@@ -0,0 +1,378 @@
+"""扩展服务生命周期、中间件排序与真实图装配。"""
+import asyncio
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+
+import pytest
+from langchain.agents.middleware.types import AgentMiddleware
+from langchain_core.messages import AIMessage
+
+import config
+from agents import agent_factory, agent_memory, agent_registry
+from db import db_conn, chat_store
+from extensions import ExtensionRegistry, get_service
+from extensions import registry as registry_module
+from extensions.services import ServiceContainer, service_view
+from runtime import environment
+from runtime.environment import RuntimeEnvironment
+from runtime.execution import execute
+from tools import registry as tools
+
+
+@asynccontextmanager
+async def resource(dependencies):
+    yield object()
+
+
+def middleware(name):
+    return lambda services: type(name, (AgentMiddleware,), {})()
+
+
+@pytest.fixture
+def isolated(monkeypatch, tmp_path):
+    schema = config.ROOT / 'db' / 'schema_sqlite.sql'
+    adapter = db_conn._SqliteAdapter(str(tmp_path / 'business.sqlite'))
+    monkeypatch.setattr(adapter, 'schema_file', lambda: schema)
+    monkeypatch.setattr(db_conn, '_adapter', adapter)
+    monkeypatch.setattr(db_conn, '_schema_ready', False)
+    monkeypatch.setattr(config, 'ROOT', tmp_path)
+    monkeypatch.setattr(agent_memory, '_saver', None)
+    monkeypatch.setattr(agent_memory, '_kind', 'memory')
+    yield
+    db_conn.close()
+
+
+def use_catalog(monkeypatch, catalog):
+    catalog.validate()
+    monkeypatch.setattr(environment, 'load_extensions', lambda: catalog)
+    monkeypatch.setattr(environment, 'approval_names', lambda: set())
+
+
+@pytest.mark.parametrize('kind', ['service', 'middleware', 'tool'])
+def test_missing_service_dependencies(kind):
+    catalog = ExtensionRegistry()
+    if kind == 'service':
+        catalog.register_service('retriever', resource, requires=['missing'])
+    elif kind == 'middleware':
+        catalog.register_middleware('Recall', middleware('Recall'), requires=['missing'])
+    else:
+        def search(query: str) -> str:
+            """检索。"""
+            return query
+        catalog.register_tool(search, requires=['missing'])
+    with pytest.raises(ValueError, match='未注册'):
+        catalog.validate()
+
+
+@pytest.mark.parametrize('kind', ['service', 'middleware'])
+def test_cycles_and_duplicate_names(kind):
+    catalog = ExtensionRegistry()
+    if kind == 'service':
+        catalog.register_service('one', resource, requires=['two'])
+        catalog.register_service('two', resource, requires=['one'])
+        with pytest.raises(ValueError, match='冲突'):
+            catalog.register_service('one', resource)
+    else:
+        catalog.register_middleware('One', middleware('One'), after=['Two'])
+        catalog.register_middleware('Two', middleware('Two'), after=['One'])
+        with pytest.raises(ValueError, match='冲突'):
+            catalog.register_middleware('One', middleware('One'))
+    with pytest.raises(ValueError, match='循环'):
+        catalog.validate()
+
+
+def test_middleware_order_scope_and_independent_instances():
+    catalog = ExtensionRegistry()
+    catalog.register_agent(name='researcher', description='检索', prompt='检索', tools=[])
+    catalog.register_middleware('Recall', middleware('Recall'), targets=['main', 'researcher'], before=['Budget'])
+    catalog.register_middleware('Budget', middleware('Budget'), targets=['main'])
+    catalog.register_middleware('ChildOnly', middleware('ChildOnly'), targets=['researcher'], after=['Recall'])
+    catalog.validate()
+    main = catalog.build_middleware('main')
+    child = catalog.build_middleware('researcher')
+    assert [m.name for m in main] == ['Recall', 'Budget']
+    assert [m.name for m in child] == ['Recall', 'ChildOnly']
+    assert main[0] is not child[0]
+    assert main[0] is not catalog.build_middleware('main')[0]
+    assert catalog.build_middleware('general-purpose') == []
+
+
+@pytest.mark.parametrize('kwargs,match', [
+    ({'targets': ['missing']}, '未注册目标'),
+    ({'after': ['missing']}, '未注册项'),
+])
+def test_invalid_middleware_references(kwargs, match):
+    catalog = ExtensionRegistry()
+    catalog.register_middleware('Recall', middleware('Recall'), **kwargs)
+    with pytest.raises(ValueError, match=match):
+        catalog.validate()
+
+
+def test_reserved_names_and_actual_middleware_name():
+    catalog = ExtensionRegistry()
+    with pytest.raises(ValueError, match='冲突'):
+        catalog.register_middleware('SummarizationMiddleware', middleware('SummarizationMiddleware'))
+    catalog.register_middleware('Recall', middleware('WrongName'))
+    with pytest.raises(ValueError, match='同名'):
+        catalog.build_middleware('main')
+
+
+@pytest.mark.asyncio
+async def test_service_order_shared_lookup_and_reverse_cleanup(isolated, monkeypatch):
+    calls = []
+    catalog = ExtensionRegistry()
+    @asynccontextmanager
+    async def database(deps):
+        assert not deps
+        value = object()
+        calls.append('open_db')
+        try:
+            yield value
+        finally:
+            calls.append('close_db')
+    @asynccontextmanager
+    async def search(deps):
+        calls.append('open_search')
+        assert set(deps) == {'db'}
+        with pytest.raises(TypeError):
+            deps['other'] = object()
+        try:
+            yield SimpleNamespace(database=deps['db'])
+        finally:
+            calls.append('close_search')
+    catalog.register_service('search', search, requires=['db'])
+    catalog.register_service('db', database)
+    use_catalog(monkeypatch, catalog)
+    old = None
+    for _ in range(2):
+        async with RuntimeEnvironment(prewarm=False, billing_retry=False):
+            assert get_service('search').database is get_service('db')
+            assert get_service('db') is not old
+            old = get_service('db')
+        with pytest.raises(RuntimeError, match='尚未启动'):
+            get_service('db')
+    assert calls == ['open_db', 'open_search', 'close_search', 'close_db'] * 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('fail_at', ['service', 'middleware'])
+async def test_startup_failure_releases_all_services(isolated, monkeypatch, fail_at):
+    calls = []
+    catalog = ExtensionRegistry()
+    @asynccontextmanager
+    async def db(deps):
+        calls.append('open')
+        try:
+            yield object()
+        finally:
+            calls.append('close')
+    @asynccontextmanager
+    async def broken(deps):
+        raise ValueError('unavailable')
+        yield
+    catalog.register_service('db', db)
+    if fail_at == 'service':
+        catalog.register_service('broken', broken, requires=['db'])
+    else:
+        catalog.register_middleware('Recall', lambda deps: object(), requires=['db'])
+    use_catalog(monkeypatch, catalog)
+    with pytest.raises(ValueError):
+        async with RuntimeEnvironment(prewarm=False, billing_retry=False):
+            pytest.fail('不可进入运行体')
+    assert calls == ['open', 'close']
+    assert not agent_memory.is_initialized()
+    assert db_conn._adapter._conn is None
+    with pytest.raises(RuntimeError, match='尚未启动'):
+        get_service('db')
+    use_catalog(monkeypatch, ExtensionRegistry())
+    async with RuntimeEnvironment(prewarm=False, billing_retry=False):
+        pass
+
+
+@pytest.mark.asyncio
+async def test_cancel_during_service_start_waits_and_closes(isolated, monkeypatch):
+    calls = []
+    started, finish = asyncio.Event(), asyncio.Event()
+    catalog = ExtensionRegistry()
+    @asynccontextmanager
+    async def slow(deps):
+        started.set()
+        await finish.wait()
+        calls.append('open')
+        try:
+            yield object()
+        finally:
+            calls.append('close')
+    catalog.register_service('slow', slow)
+    use_catalog(monkeypatch, catalog)
+    async def run():
+        async with RuntimeEnvironment(prewarm=False, billing_retry=False):
+            pytest.fail('取消后不可进入运行体')
+    task = asyncio.create_task(run())
+    await started.wait()
+    task.cancel()
+    await asyncio.sleep(0)
+    task.cancel()
+    finish.set()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert calls == ['open', 'close']
+    with pytest.raises(RuntimeError):
+        get_service('slow')
+
+
+def test_disabled_extension_skips_config_validation_and_setup(monkeypatch):
+    def forbidden(*args):
+        pytest.fail('停用模块不能执行校验或 setup')
+    fake = SimpleNamespace(enabled=lambda: False, validate_config=forbidden, setup=forbidden)
+    original = registry_module.importlib.import_module
+    monkeypatch.setattr(registry_module, 'discover_modules', lambda: ['extensions.core.extension', 'optional'])
+    monkeypatch.setattr(registry_module.importlib, 'import_module', lambda name: fake if name == 'optional' else original(name))
+    tools.reset_for_tests()
+    catalog = registry_module.load_extensions()
+    assert catalog.disabled_modules == ['optional']
+    assert catalog.modules == ['extensions.core.extension']
+
+
+def test_extension_config_failure_prevents_publication(monkeypatch):
+    def invalid():
+        raise ValueError('missing endpoint')
+    fake = SimpleNamespace(enabled=lambda: True, validate_config=invalid,
+                           setup=lambda catalog: pytest.fail('配置错误后不能注册'))
+    monkeypatch.setattr(registry_module, 'discover_modules', lambda: ['broken'])
+    monkeypatch.setattr(registry_module.importlib, 'import_module', lambda name: fake)
+    tools.reset_for_tests()
+    with pytest.raises(ValueError, match='missing endpoint'):
+        registry_module.load_extensions()
+    assert registry_module._loaded is None
+    assert not tools.get_tool_names()
+
+
+@pytest.mark.asyncio
+async def test_enabled_demo_runs_real_graph_without_network(isolated, monkeypatch):
+    from test_runtime_regressions import ScriptedModel
+    from extensions.context_demo.extension import lookup_demo_term
+    from pydantic import Field
+    class Model(ScriptedModel):
+        seen: list[str] = Field(default_factory=list)
+        def _stream(self, messages, *args, **kwargs):
+            self.seen.append(str(messages[0].content))
+            yield from super()._stream(messages, *args, **kwargs)
+    monkeypatch.setenv('CONTEXT_DEMO_ENABLED', '1')
+    tools.reset_for_tests()
+    model = Model(responses=[
+        AIMessage(content='', tool_calls=[{'id': 'term', 'name': 'lookup_demo_term', 'args': {'term': '风量'}}]),
+        AIMessage(content='已查询术语表。'),
+    ])
+    monkeypatch.setattr('agents.model_factory.get_model', lambda: model)
+    async with RuntimeEnvironment(prewarm=False, billing_retry=False):
+        assert '空气体积' in lookup_demo_term('风量')
+        sid = chat_store.create_session('demo-user')['id']
+        result = await execute(user_name='demo-user', session_id=sid, message='什么是风量？')
+        assert result.status == 'success'
+        assert '已查询术语表' in result.output
+        assert model.seen and all('演示术语表可查询：风速、风量' in text for text in model.seen)
+        assert model.seen[0].count('演示术语表可查询') == 1
+        assert model.seen[1].count('演示术语表可查询') == 1
+    assert agent_registry.peek_agent() is None
+    with pytest.raises(RuntimeError, match='尚未启动'):
+        lookup_demo_term('风量')
+
+
+def test_agent_factory_attaches_middleware_only_to_requested_targets(monkeypatch):
+    catalog = registry_module.load_extensions()
+    # 使用独立目录副本，保留核心工具，避免修改共享扩展缓存。
+    local = ExtensionRegistry()
+    local.tools.update(catalog.tools)
+    local.agents.update(catalog.agents)
+    local.register_middleware('ChildOnly', middleware('ChildOnly'), targets=['demo-searcher'])
+    local.register_middleware('MainOnly', middleware('MainOnly'))
+    local.validate()
+    monkeypatch.setattr(agent_factory, 'load_extensions', lambda: local)
+    monkeypatch.setattr(agent_factory.model_factory, 'get_model', lambda: object())
+    monkeypatch.setattr(agent_factory, '_build_backend', lambda: None)
+    monkeypatch.setattr('deepagents.create_deep_agent', lambda **kwargs: kwargs)
+    agent = agent_factory.build_agent()
+    assert [m.name for m in agent['middleware']] == ['DynamicTimeMiddleware', 'MainOnly']
+    assert [m.name for m in agent['subagents'][0]['middleware']] == ['ChildOnly']
+
+
+@pytest.mark.asyncio
+async def test_cleanup_failure_does_not_skip_other_services(isolated, monkeypatch):
+    calls = []
+    catalog = ExtensionRegistry()
+    @asynccontextmanager
+    async def db(deps):
+        try:
+            yield object()
+        finally:
+            calls.append('db_closed')
+    @asynccontextmanager
+    async def search(deps):
+        try:
+            yield object()
+        finally:
+            calls.append('search_closed')
+            raise RuntimeError('close failed')
+    catalog.register_service('db', db)
+    catalog.register_service('search', search, requires=['db'])
+    use_catalog(monkeypatch, catalog)
+    with pytest.raises(RuntimeError, match='close failed'):
+        async with RuntimeEnvironment(prewarm=False, billing_retry=False):
+            pass
+    assert calls == ['search_closed', 'db_closed']
+    assert not agent_memory.is_initialized()
+    assert db_conn._adapter._conn is None
+    with pytest.raises(RuntimeError):
+        get_service('db')
+
+
+@pytest.mark.asyncio
+async def test_middleware_executes_only_on_target_in_real_delegation(isolated, monkeypatch):
+    from test_runtime_regressions import ScriptedModel
+    seen = []
+    def factory(name):
+        class Middleware(AgentMiddleware):
+            async def awrap_model_call(self, request, handler):
+                seen.append(name + '_before')
+                result = await handler(request)
+                seen.append(name + '_after')
+                return result
+        Middleware.__name__ = name
+        return lambda deps: Middleware()
+    base = registry_module.load_extensions()
+    local = ExtensionRegistry()
+    local.tools.update(base.tools)
+    local.agents.update(base.agents)
+    local.register_middleware('MainOuter', factory('MainOuter'), before=['MainInner'])
+    local.register_middleware('MainInner', factory('MainInner'))
+    local.register_middleware('ChildOnly', factory('ChildOnly'), targets=['demo-searcher'])
+    local.validate()
+    monkeypatch.setattr(registry_module, '_loaded', local)
+    model = ScriptedModel(responses=[
+        AIMessage(content='', tool_calls=[{'id': 'delegate', 'name': 'task',
+                  'args': {'subagent_type': 'demo-searcher', 'description': '查询术语'}}]),
+        AIMessage(content='检索结论。'), AIMessage(content='汇总完成。'),
+    ])
+    monkeypatch.setattr('agents.model_factory.get_model', lambda: model)
+    async with RuntimeEnvironment(prewarm=False, billing_retry=False):
+        sid = chat_store.create_session('delegation')['id']
+        result = await execute(user_name='delegation', session_id=sid, message='请检索并汇总')
+        assert result.status == 'success', result.error
+        assert result.output.endswith('汇总完成。')
+    main_call = ['MainOuter_before', 'MainInner_before', 'MainInner_after', 'MainOuter_after']
+    assert seen == main_call + ['ChildOnly_before', 'ChildOnly_after'] + main_call
+
+
+@pytest.mark.parametrize('hook', ['enabled', 'validate_config', 'setup'])
+def test_async_declaration_hooks_are_rejected(monkeypatch, hook):
+    async def asynchronous(*args):
+        return True
+    fake = SimpleNamespace(enabled=lambda: True, validate_config=lambda: None, setup=lambda registry: None)
+    setattr(fake, hook, asynchronous)
+    monkeypatch.setattr(registry_module, 'discover_modules', lambda: ['async_extension'])
+    monkeypatch.setattr(registry_module.importlib, 'import_module', lambda name: fake)
+    tools.reset_for_tests()
+    with pytest.raises(ValueError, match='同步'):
+        registry_module.load_extensions()
+    assert registry_module._loaded is None