```

feat(agent): 添加对持久化子智能体的支持并增强委派管理

添加了持久化子智能体的完整生命周期管理功能，包括创建、更新、删除和查询API接口。
新增了子智能体的JSON-RPC通信协议支持，实现了远程调用和任务管理功能。

同时增强了委派管理器的功能：
- 添加了对本地委派、插件委派和本地回退的开关控制
- 实现了持久化子智能体任务的自动检测和本地执行保护
- 增加了对不同委派类型的权限验证机制

修改了智能体注册表以支持插件智能体的条件性包含，并更新了工具注册逻辑以支持可选工具。

BREAKING CHANGE: 委派管理器的构造函数签名已更改，添加了新的控制参数。
```

app-instance/backend/nanobot/agent/delegation.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import asyncio
+import re
 import uuid
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
@@ -68,6 +69,9 @@ class DelegationManager:
         allowed_hosts: list[str] | None = None,
         authz_config: Any | None = None,
         backend_identity: Any | None = None,
+        allow_local_delegation: bool = True,
+        allow_plugin_delegation: bool = True,
+        allow_local_fallback: bool = True,
     ):
         self.provider = provider
         self.workspace = workspace
@@ -76,6 +80,9 @@ class DelegationManager:
         # local_executor 只负责“本地执行”，不再承担队列编排职责。
         self.local_executor = local_executor
         self.max_parallel_agents = max(1, max_parallel_agents)
+        self.allow_local_delegation = allow_local_delegation
+        self.allow_plugin_delegation = allow_plugin_delegation
+        self.allow_local_fallback = allow_local_fallback
         # A2AClient 只处理远端协议细节，委派策略和公告统一放在本类。
         self.a2a_client = A2AClient(
             timeout_seconds=timeout_seconds,
@@ -88,6 +95,19 @@ class DelegationManager:
         self._running_tasks: dict[str, DelegationRun] = {}
         self._direct_announcement_callback: DirectAnnouncementCallback | None = None
 
+    _PERSISTENT_SUBAGENT_PATTERNS = (
+        re.compile(r"\bsub[\s-]?agent\b", re.IGNORECASE),
+        re.compile(r"\bpersistent\b", re.IGNORECASE),
+        re.compile(r"\bagents\.json\b", re.IGNORECASE),
+        re.compile(r"\bregistry\.json\b", re.IGNORECASE),
+        re.compile(r"\bsubagentctl\b", re.IGNORECASE),
+        re.compile(r"/api/subagents", re.IGNORECASE),
+        re.compile(r"workspace/agents", re.IGNORECASE),
+        re.compile(r"子智能体"),
+        re.compile(r"子 agent", re.IGNORECASE),
+        re.compile(r"持久化"),
+    )
+
     def set_direct_announcement_callback(
         self,
         callback: DirectAnnouncementCallback | None,
@@ -160,6 +180,7 @@ class DelegationManager:
         label: str,
         *,
         parent_run_id: str | None = None,
+        task: str | None = None,
     ) -> None:
         # 单 agent 执行开始事件，供前端画执行树。
         await emit_process_event(
@@ -177,8 +198,21 @@ class DelegationManager:
                 "protocol": descriptor.protocol,
                 "support_group": descriptor.support_group,
                 "support_streaming": descriptor.support_streaming,
+                "delegated_task": task,
             },
         )
+        if task:
+            await emit_process_event(
+                "process_run_message",
+                run_id=run_id,
+                parent_run_id=parent_run_id,
+                actor_type="agent",
+                actor_id=descriptor.id,
+                actor_name=descriptor.name,
+                message_role="user",
+                text=task,
+                metadata={"source": "delegation_input"},
+            )
 
     async def _emit_agent_finished(
         self,
@@ -386,7 +420,7 @@ class DelegationManager:
 
             # 单 agent 场景先解析目标，再执行。
             descriptor = self._resolve_single(task, target, strategy)
-            await self._emit_agent_started(run_id, descriptor, label)
+            await self._emit_agent_started(run_id, descriptor, label, task=task)
             progress_callback = self._build_progress_callback(
                 origin,
                 descriptor,
@@ -468,15 +502,20 @@ class DelegationManager:
             descriptor = self.registry.get_agent(target)
             if descriptor is None:
                 raise ValueError(f"Agent '{target}' not found")
+            self._ensure_descriptor_allowed(descriptor)
             return descriptor
 
         if strategy == "local":
+            if not self.allow_local_fallback:
+                raise ValueError("Local fallback delegation is disabled")
             descriptor = self.registry.get_agent("local-subagent")
             if descriptor is None:
                 raise ValueError("Local subagent is not available")
             return descriptor
 
         if strategy == "plugin":
+            if not self.allow_plugin_delegation:
+                raise ValueError("Plugin delegation is disabled")
             suggestions = [
                 agent for agent in self.registry.suggest_agents(task)
                 if agent.kind == "local_prompt" and agent.source == "plugin"
@@ -494,15 +533,46 @@ class DelegationManager:
                 return suggestions[0]
             raise ValueError("No matching A2A agent found")
 
-        suggestions = self.registry.suggest_agents(task, limit=1)
+        # Persistent sub-agent 管理是本地工作区变更任务，必须留在本地执行，
+        # 不能自动委派给远端 A2A agent，否则远端看不到本地规范和状态。
+        if self._is_persistent_subagent_task(task):
+            if not self.allow_local_fallback:
+                raise ValueError("Persistent sub-agent management requires local fallback delegation")
+            descriptor = self.registry.get_agent("local-subagent")
+            if descriptor is None:
+                raise ValueError("Local fallback agent is not available")
+            return descriptor
+
+        suggestions = [
+            agent for agent in self.registry.suggest_agents(task, limit=5)
+            if self._descriptor_allowed(agent)
+        ]
         if suggestions:
             return suggestions[0]
         # 自动路由一个都猜不到时，最后回到本地兜底 agent。
+        if not self.allow_local_fallback:
+            raise ValueError("No allowed agent found for delegation")
         descriptor = self.registry.get_agent("local-subagent")
         if descriptor is None:
             raise ValueError("Local fallback agent is not available")
         return descriptor
 
+    @classmethod
+    def _is_persistent_subagent_task(cls, task: str) -> bool:
+        text = (task or "").strip()
+        if not text:
+            return False
+
+        matched = sum(1 for pattern in cls._PERSISTENT_SUBAGENT_PATTERNS if pattern.search(text))
+        if matched >= 2:
+            return True
+
+        lowered = text.lower()
+        return (
+            ("create" in lowered or "update" in lowered or "repair" in lowered or "fix" in lowered)
+            and ("subagent" in lowered or "sub-agent" in lowered)
+        )
+
     async def _run_group(
         self,
         task: str,
@@ -520,7 +590,10 @@ class DelegationManager:
             resolved_targets.append(target)
         if not resolved_targets:
             # 未显式给出目标时，根据任务文本自动挑若干个候选 agent。
-            suggestions = self.registry.suggest_agents(task, limit=self.max_parallel_agents)
+            suggestions = [
+                agent for agent in self.registry.suggest_agents(task, limit=self.max_parallel_agents * 2)
+                if self._descriptor_allowed(agent)
+            ]
             resolved_targets = [agent.id for agent in suggestions]
         if not resolved_targets:
             raise ValueError("No agents available for group delegation")
@@ -533,6 +606,7 @@ class DelegationManager:
             if descriptor is None:
                 missing.append(item)
             else:
+                self._ensure_descriptor_allowed(descriptor)
                 descriptors.append(descriptor)
         if missing:
             raise ValueError(f"Agent(s) not found: {', '.join(missing)}")
@@ -544,7 +618,13 @@ class DelegationManager:
             child_run_id = new_run_id("agent")
             async with semaphore:
                 try:
-                    await self._emit_agent_started(child_run_id, descriptor, label, parent_run_id=run_id)
+                    await self._emit_agent_started(
+                        child_run_id,
+                        descriptor,
+                        label,
+                        parent_run_id=run_id,
+                        task=task,
+                    )
                     result = await self._execute_descriptor(
                         descriptor,
                         task,
@@ -588,6 +668,12 @@ class DelegationManager:
         """根据 descriptor 类型执行具体 agent。"""
         logger.info("Delegating '{}' to {}", label, descriptor.id)
         if descriptor.kind in {"local_fallback", "local_prompt"}:
+            if not self.allow_local_delegation or (
+                descriptor.kind == "local_prompt" and not self.allow_plugin_delegation
+            ) or (
+                descriptor.kind == "local_fallback" and not self.allow_local_fallback
+            ):
+                raise ValueError(f"Delegation to '{descriptor.id}' is disabled")
             # 本地执行时，把当前 run_id 写入上下文，便于更深层的 MCP/tool 事件挂父节点。
             with process_run_context(process_run_id):
                 return await self.local_executor.run_local_task(
@@ -611,6 +697,19 @@ class DelegationManager:
                 )
         raise ValueError(f"Unsupported agent kind '{descriptor.kind}'")
 
+    def _descriptor_allowed(self, descriptor: AgentDescriptor) -> bool:
+        if descriptor.kind == "local_fallback":
+            return self.allow_local_fallback and self.allow_local_delegation
+        if descriptor.kind == "local_prompt":
+            return self.allow_local_delegation and self.allow_plugin_delegation
+        if descriptor.protocol == "a2a" or descriptor.kind == "a2a_remote":
+            return True
+        return False
+
+    def _ensure_descriptor_allowed(self, descriptor: AgentDescriptor) -> None:
+        if not self._descriptor_allowed(descriptor):
+            raise ValueError(f"Delegation to '{descriptor.id}' is disabled")
+
     def _build_progress_callback(
         self,
         origin: dict[str, str],