[Feature]: In-process sub-agent execution & EventBase metadata for event routing

**Title (EN)**: `[Feature]: In-process sub-agent execution & EventBase metadata for event routing`

**标题（中文）**: `[Feature]: 进程内 sub-agent 执行与 EventBase metadata 事件路由`

> English version first, 中文版在文末

Related to:
- #4622
- #4749
---

# English

## Summary

Two related feature requests:

1. **`EventBase.metadata`** — add a generic `metadata: dict[str, Any]` field to `agentscope.event.EventBase`, enabling downstream consumers to attach routing/context information to events without subclassing.
2. **In-process sub-agent execution** — enhance the existing `spawn_subagent` tool so that ephemeral sub-agents can run in-process (sharing the parent agent's event stream), rather than only via HTTP round-trip to the runner.

These are independent proposals but synergize: in-process sub-agents produce events on the same stream as the parent, and `metadata` is the minimal protocol change that lets consumers (frontends, channels) distinguish which sub-agent produced which event.

## Component(s) Affected

- [x] Core / Backend — agentscope `EventBase`, agent tool execution, event stream
- [ ] Console (frontend web UI)
- [ ] Channels
- [ ] Skills
- [ ] CLI
- [ ] Documentation
- [ ] Tests
- [ ] CI/CD

## Problem / Motivation

### 1. No metadata on events

`agentscope.event.EventBase` currently has only `id` and `created_at`:

```python
class EventBase(BaseModel):
id: str = Field(default_factory=lambda: uuid.uuid4().hex)
created_at: str = Field(default_factory=lambda: datetime.now().isoformat())
```

When multiple agents (or sub-agents) produce events on the same stream, consumers have no standard way to distinguish event sources. Applications resort to:

- Post-processing serialized JSON to inject fields (fragile, doesn't round-trip through `model_validate`)
- Subclassing every event type (doesn't compose across plugins)
- Building per-stream lookup tables to correlate events by `reply_id` (complex, error-prone)

### 2. `spawn_subagent` is HTTP-only

QwenPaw's current `spawn_subagent` (`agents/tools/agent_management.py:646`) works by making HTTP calls to the runner API:

```python
async def spawn_subagent(task, fork=False, background=False, timeout=600):
# ... builds request_payload ...
response_data = await asyncio.to_thread(
collect_final_agent_chat_response,
None, request_payload, current_agent_id, timeout,
)
```

This means:
- Sub-agent events are invisible to the parent's event stream — they go through a separate HTTP session
- No way for frontends to show sub-agent progress inline with the parent agent's output
- Parallel sub-agents each create independent HTTP sessions; their events cannot be multiplexed onto one stream
- The HTTP round-trip adds latency and serialization overhead for what could be an in-process call

For workflows where a parent agent needs to dispatch multiple independent sub-tasks in parallel and show their progress in a unified view, in-process execution with event-stream multiplexing is needed.

### Example: DataPaw DAG parallel execution

DataPaw's DAG-based analysis agent illustrates both needs. A DAG like:

```
[取数 A] ──┐
├──→ [合并分析]
[取数 B] ──┘
```

Today executes strictly serially (one node per reasoning loop). With in-process sub-agents:

1. Master agent sees nodes "取数 A" and "取数 B" are ready (no dependencies)
2. Master calls `spawn_subagent(node_id="n1", task="取数 A")` and `spawn_subagent(node_id="n2", task="取数 B")` in the same tool-call turn
3. agentscope's existing `_execute_tool_calls_concurrently` runs both via `asyncio.gather`
4. Both sub-agents produce events tagged with `metadata = {"node_id": "n1", "subagent_id": "sa_xxx"}` / `{"node_id": "n2", "subagent_id": "sa_yyy"}`
5. Frontend routes events to the correct DAG node panel by reading `metadata`

Without `EventBase.metadata`, step 4 requires fragile JSON post-processing. Without in-process execution, step 3 is impossible — each sub-agent goes through a separate HTTP session.

A single DAG node may also spawn multiple sub-agents (e.g., "compare dataset A vs B" spawns one sub-agent per dataset). In this case `node_id` alone is insufficient — `subagent_id` provides instance-level distinction.

## Proposed Solution

### Proposal 1: `EventBase.metadata`

Add one field to `agentscope.event.EventBase`:

```python
class EventBase(BaseModel):
model_config = ConfigDict(use_enum_values=True)

id: str = Field(default_factory=lambda: uuid.uuid4().hex)
created_at: str = Field(default_factory=lambda: datetime.now().isoformat())
metadata: dict[str, Any] = Field(default_factory=dict) # NEW
```

**Properties:**

- Backward compatible: defaults to `{}`, no behavioral change for existing code
- Generic: not tied to any specific application — plugins, channels, frontends all benefit
- All existing event subclasses (`ReplyStartEvent`, `ToolCallStartEvent`, etc.) inherit the field automatically

**Use cases beyond DataPaw:**

| Use case | metadata keys |
|----------|--------------|
| Multi-tenant platforms | `tenant_id`, `user_id` |
| Observability / tracing | `trace_id`, `span_id` |
| Plugin event routing | `plugin_id`, `source` |
| A/B testing | `experiment_id`, `variant` |
| DAG node routing (DataPaw) | `graph_id`, `node_id`, `subagent_id` |

**Relationship to the existing plugin API issue (T1/T2):**

The existing DataPaw plugin API issue requests `metadata` on QwenPaw's runtime-level `Event` schema (`agentscope_runtime.engine.schemas.agent_schemas.Event`) plus Msg→Event propagation. This proposal is complementary — it adds `metadata` at the agentscope framework level (`EventBase`), which is upstream of both QwenPaw's runtime Event and the Msg→Event converter. Both changes are needed:

- `EventBase.metadata` — framework level, carries metadata through the agent's internal event pipeline
- Runtime `Event.metadata` + Msg→Event propagation (T1 in existing issue) — runtime level, carries metadata to the wire/SSE layer

### Proposal 2: In-process sub-agent execution

Enhance `spawn_subagent` to support in-process execution. The key insight: agentscope already has `_execute_tool_calls_concurrently` which runs multiple tool calls in parallel via `asyncio.gather`. A `spawn_subagent` tool that internally creates an ephemeral `Agent` instance gets parallelism for free from this existing infrastructure.

**Sub-agent characteristics:**

| Aspect | Master agent | Sub-agent |
|--------|-------------|-----------|
| Lifecycle | Session-scoped | Ephemeral, destroyed after task |
| Toolkit | Full (orchestration + execution) | Execution only |
| System prompt | Full | Minimal task-specific prompt |
| State | Owns `AgentState` | None |
| Memory | Full conversation history | Task description + context only |
| Events | `metadata = {}` | `metadata = {subagent_id: "..."}` |

**Tool schema:**

```python
async def spawn_subagent(
task: str,
context: str | None = None,
# existing params preserved for backward compat:
fork: bool = False,
background: bool = False,
timeout: int = 600,
) -> ToolResponse:
"""Spawn an ephemeral sub-agent to execute a specific task.

The sub-agent runs in-process with its own reasoning loop.
Call multiple times in the same turn to run tasks in parallel.

Args:
task: The task for the sub-agent to execute.
context: Optional context (upstream results, constraints, etc.).
fork: If True, inherits parent session state (existing behavior).
background: If True, runs via HTTP as today (existing behavior).
timeout: Execution timeout in seconds.
"""
```

When neither `fork` nor `background` is set, the tool runs the sub-agent in-process. This is backward compatible — existing `fork=True` and `background=True` paths are unchanged.

**Streaming events vs final return — two layers:**

A sub-agent is a ReAct agent: during execution it continuously produces streaming events (thinking, tool calls, text chunks). There are two consumers with different needs:

| Layer | Carrier | Consumer | Purpose |
|-------|---------|----------|---------|
| During execution | `EventBase` (SSE push) | Frontend | Real-time display of sub-agent's thinking, tool calls, text output |
| After execution | `ToolResponse` (tool return value) | Master agent's LLM | Master agent decides next step based on this |

**Prerequisite: extend tool execution pipeline to support `EventBase` passthrough**

Currently a tool can only yield `ToolChunk`/`ToolResponse`. Sub-agent events (`ThinkingBlockDeltaEvent`, `ToolCallStartEvent`, `TextBlockDeltaEvent`, etc.) are `EventBase` subclasses — they have no way to reach `reply_stream()`. Three changes to agentscope are needed:

1. **`toolkit.call_tool`** (`agentscope/tool/_toolkit.py:321-324`): when iterating a tool's async generator output, if the yielded object is `EventBase`, passthrough yield it without accumulating into `ToolResponse`:

```python
elif isinstance(res, AsyncGenerator):
async for chunk in res:
if isinstance(chunk, EventBase):
yield chunk # passthrough, don't accumulate
else:
yield chunk
tool_response.append_chunk(chunk)
```

2. **`_execute_tool_call`** (`agentscope/agent/_agent.py:1380-1466`): when iterating `_acting()`, if the chunk is `EventBase`, yield it directly to `reply_stream()` — don't convert via `_convert_tool_chunk_to_event`:

```python
async for chunk in self._acting(tool_call):
if isinstance(chunk, ToolResponse):
# existing: save to context, yield ToolResultEndEvent
...
elif isinstance(chunk, EventBase):
# NEW: passthrough sub-agent events
yield chunk
else:
# existing: convert ToolChunk to ToolResultTextDeltaEvent etc.
...
```

3. **Type signatures**: `_acting`, `_acting_impl`, `call_tool` yield types expand from `ToolChunk | ToolResponse` to `ToolChunk | ToolResponse | EventBase`.

This is a generic extension — any tool that internally runs an agent (or any event-producing subsystem) benefits. The `spawn_subagent` tool uses it by iterating `sub_agent.reply_stream()`, injecting `metadata` on each event, and yielding them as `EventBase` objects.

**Streaming events (frontend consumption):**

Every `EventBase` produced by a sub-agent (`ReplyChunkEvent`, `ToolCallStartEvent`, etc.) is passthrough-yielded by the tool and carries metadata:

```python
metadata = {"subagent_id": "sa_7kX2m"}
```

Frontend routes by `metadata`: has `subagent_id` → display in side panel (e.g., task/DAG node panel); absent → display in chat panel. Applications can add their own keys (e.g., DataPaw adds `graph_id` + `node_id` for DAG node routing).

**Final return (master agent consumption):**

After the sub-agent finishes, `spawn_subagent` yields a final `ToolChunk` as the tool result for the master agent's LLM. The accumulated `ToolResponse` leverages its existing fields:

```python
ToolResponse(
state=ToolResultState.SUCCESS, # or ERROR
content=[TextBlock(text="...")], # sub-agent's final text output
metadata={ # structured result for programmatic access
"subagent_id": "sa_7kX2m",
"files": [ # artifacts produced (if any)
{"name": "report.csv", "path": "/workspace/artifacts/report.csv", "mime_type": "text/csv"},
],
},
)
```

| field | contents |
|-------|----------|
| `state` | `SUCCESS` if sub-agent completed normally; `ERROR` if it failed or timed out |
| `content` | Sub-agent's final text output as `TextBlock` — the analysis result, data summary, error description, etc. This is what the master agent's LLM reads |
| `metadata` | `subagent_id` (always present), `files` (list of produced artifacts, optional). Applications can add domain-specific keys (e.g., DataPaw adds `node_id`). Master agent's tool execution layer can read `metadata` without parsing `content` text |

Error example:

```python
ToolResponse(
state=ToolResultState.ERROR,
content=[TextBlock(text="Failed to connect to database: connection refused on port 5432")],
metadata={"subagent_id": "sa_7kX2m"},
)
```

**Event flow:**

```
Master agent reply_stream()
│
├─ tool_call: spawn_subagent(task="A") ─┐
├─ tool_call: spawn_subagent(task="B") ─┤ asyncio.gather (existing infra)
│ │
│ sub-agent-1 (ReAct loop): │ sub-agent-2 (ReAct loop):
│ streaming EventBase with │ streaming EventBase with
│ metadata {subagent_id: "sa_1"} │ metadata {subagent_id: "sa_2"}
│ → frontend shows progress │ → frontend shows progress
│ │
│ final ToolResponse │ final ToolResponse
│ → master agent reads content │ → master agent reads content
│ │
└─── master agent continues reasoning ───┘
```

**Error handling:**

- One sub-agent failing does not cancel others (consistent with `_execute_tool_calls_concurrently`)
- Failed sub-agents return error text as tool results
- Master agent decides retry/skip/fail

### Tasks

- [ ] **T1**: Add `metadata: dict[str, Any]` to `agentscope.event.EventBase`
- [ ] **T2**: Extend tool execution pipeline — `toolkit.call_tool`, `_execute_tool_call`, `_acting` support `EventBase` passthrough alongside `ToolChunk`/`ToolResponse`
- [ ] **T3**: Implement in-process path for `spawn_subagent` (when `fork=False, background=False`), yielding sub-agent `EventBase` events with metadata through the pipeline
- [ ] **T4**: Document metadata conventions and sub-agent usage patterns

### Suggested landing order

1. **T1 + T2** (protocol): T1 is a one-field addition to `EventBase`. T2 extends the tool pipeline to support `EventBase` passthrough — both are prerequisites for sub-agent event routing.
2. **T3** (in-process sub-agent): Depends on T1 + T2. Implements `spawn_subagent` with in-process execution + event streaming.
3. **T4** (docs): After all land.

## Alternatives Considered

- **Keep HTTP-only `spawn_subagent`**: Works for cross-agent delegation, but sub-agent events are invisible to the parent's stream. No way to show inline progress for parallel sub-tasks.
- **Side-channel event merging** (application-level SSE merge without upstream changes): The application's SSE endpoint merges events from `reply_stream()` and a side-channel broadcaster that sub-agent tools push to. Works without upstream changes, but: (1) every application must implement its own merge logic, (2) sub-agent events bypass agentscope's event pipeline (middleware, interceptors don't see them), (3) concurrent event ordering is harder to guarantee outside the framework. The pipeline extension (Proposal 2) is more general and composable.
- **Add a `spawn_parallel(node_ids=[...])` batch tool**: Rejected — violates atomicity. The master agent should decide parallelism by issuing multiple atomic `spawn_subagent` calls, not by passing an array. This is simpler, more flexible (same node can have multiple sub-agents), and consistent with how LLMs naturally issue parallel tool calls.
- **Subclass every event type to add metadata**: Doesn't compose across plugins. Every application would define its own event subclasses, and middleware/interceptors couldn't generically access the metadata.
- **Use `reply_id` for sub-agent routing**: Insufficient — `reply_id` identifies a reply, not a sub-agent. Multiple sub-agents spawned in the same tool-call turn share the parent's `reply_id`.

## Additional Context

- Existing `spawn_subagent` implementation: `src/qwenpaw/agents/tools/agent_management.py:646-745`
- agentscope `EventBase`: `agentscope/event/_event.py:53-62`
- agentscope parallel tool execution: `agentscope/agent/_agent.py:1130-1204` (`_execute_tool_calls_concurrently`, uses `asyncio.gather`)
- DataPaw sub-agent design doc: `datapaw/design-docs/2026-06-02-sub-agent-parallel-execution.md`
- Related issue: DataPaw plugin API extensions (`datapaw-docs/2026-05-28-host-plugin-api-issue.md`, T1/T2 cover runtime-level Event metadata)

## Willing to Contribute

- [x] I am willing to open a PR for this feature (after discussion).

DataPaw maintainers will contribute the implementation and tests. We can provide concrete usage data from DAG parallel execution scenarios.

---

# 中文

## Summary

两个关联的 feature request：

1. **`EventBase.metadata`** — 给 `agentscope.event.EventBase` 增加通用的 `metadata: dict[str, Any]` 字段，让下游消费方可以在事件上附加路由/上下文信息，无需 subclass 每种事件类型。
2. **进程内 sub-agent 执行** — 增强现有 `spawn_subagent` 工具，让临时 sub-agent 可以在进程内执行（共享父 agent 的事件流），而不是只能走 HTTP round-trip 到 runner。

两者独立可落地但互相增强：进程内 sub-agent 在同一条流上产出事件，而 `metadata` 是让消费方（前端、channel）区分事件来源的最小协议变更。

## Component(s) Affected

- [x] Core / Backend — agentscope `EventBase`、agent 工具执行、事件流
- [ ] Console (frontend web UI)
- [ ] Channels
- [ ] Skills
- [ ] CLI
- [ ] Documentation
- [ ] Tests
- [ ] CI/CD

## Problem / Motivation

### 1. 事件上没有 metadata

`agentscope.event.EventBase` 目前只有 `id` 和 `created_at`：

```python
class EventBase(BaseModel):
id: str = Field(default_factory=lambda: uuid.uuid4().hex)
created_at: str = Field(default_factory=lambda: datetime.now().isoformat())
```

当多个 agent（或 sub-agent）在同一条流上产出事件时，消费方没有标准手段区分事件来源。应用只能：

- 在序列化层后处理 JSON 注入字段（脆弱，不能通过 `model_validate` round-trip）
- 给每种事件类型做 subclass（不同应用无法组合）
- 按 `reply_id` 建 per-stream 查询表做事件关联（复杂、易出错）

### 2. `spawn_subagent` 只走 HTTP

QwenPaw 当前的 `spawn_subagent`（`agents/tools/agent_management.py:646`）通过 HTTP 调用 runner API：

```python
response_data = await asyncio.to_thread(
collect_final_agent_chat_response,
None, request_payload, current_agent_id, timeout,
)
```

这意味着：
- Sub-agent 的事件对父 agent 的事件流不可见 — 走的是独立 HTTP session
- 前端无法在父 agent 输出中 inline 展示 sub-agent 进度
- 并行 sub-agent 各自创建独立 HTTP session，事件无法在一条流上复用
- HTTP round-trip 带来延迟和序列化开销 — 本可以是进程内调用

对于父 agent 需要并行分发多个独立子任务、并在统一视图中展示进度的场景，需要进程内执行 + 事件流复用。

### 示例：DataPaw DAG 并行执行

DataPaw 的 DAG 数据分析 agent 同时体现两个需求。一个 DAG：

```
[取数 A] ──┐
├──→ [合并分析]
[取数 B] ──┘
```

今天严格串行（一轮 reasoning 只处理一个 node）。有了进程内 sub-agent：

1. 主 agent 发现 "取数 A" 和 "取数 B" 都 ready（无依赖）
2. 主 agent 在同一个 tool-call turn 调 `spawn_subagent(task="取数 A")` 和 `spawn_subagent(task="取数 B")`
3. agentscope 已有的 `_execute_tool_calls_concurrently` 通过 `asyncio.gather` 并行执行
4. 两个 sub-agent 的事件都带 `metadata = {"node_id": "n1", "subagent_id": "sa_xxx"}` / `{"node_id": "n2", "subagent_id": "sa_yyy"}`
5. 前端读 `metadata` 路由到对应 DAG 节点面板

没有 `EventBase.metadata`，第 4 步需要脆弱的 JSON 后处理。没有进程内执行，第 3 步不可能 — 每个 sub-agent 走独立 HTTP session。

单个 DAG 节点也可能 spawn 多个 sub-agent（如 "对比 A/B 数据集" 每个数据集一个 sub-agent）。此时仅靠 `node_id` 不够 — `subagent_id` 提供实例级区分。

## Proposed Solution

### Proposal 1: `EventBase.metadata`

给 `agentscope.event.EventBase` 加一个字段：

```python
class EventBase(BaseModel):
model_config = ConfigDict(use_enum_values=True)

id: str = Field(default_factory=lambda: uuid.uuid4().hex)
created_at: str = Field(default_factory=lambda: datetime.now().isoformat())
metadata: dict[str, Any] = Field(default_factory=dict) # 新增
```

**属性：**

- 向后兼容：默认 `{}`，对现有代码零行为变更
- 通用：不绑定特定应用 — plugin、channel、前端都受益
- 所有已有事件子类（`ReplyStartEvent`、`ToolCallStartEvent` 等）自动继承

**DataPaw 之外的用例：**

| 用例 | metadata keys |
|------|--------------|
| 多租户平台 | `tenant_id`, `user_id` |
| 可观测性 / 链路追踪 | `trace_id`, `span_id` |
| 插件事件路由 | `plugin_id`, `source` |
| A/B 测试 | `experiment_id`, `variant` |
| DAG 节点路由（DataPaw） | `graph_id`, `node_id`, `subagent_id` |

**与现有 plugin API issue（T1/T2）的关系：**

现有 DataPaw plugin API issue 请求在 QwenPaw 运行时层面的 `Event` schema（`agentscope_runtime.engine.schemas.agent_schemas.Event`）加 `metadata` + Msg→Event propagation。本提案是互补的 — 在 agentscope 框架层面（`EventBase`）加 `metadata`，位于 QwenPaw 的 runtime Event 和 Msg→Event 转换器的上游。两个变更都需要：

- `EventBase.metadata` — 框架层，metadata 在 agent 内部事件管道中流转
- Runtime `Event.metadata` + Msg→Event propagation（现有 issue T1）— 运行时层，metadata 到达 wire/SSE 层

### Proposal 2: 进程内 sub-agent 执行

增强 `spawn_subagent` 支持进程内执行。关键洞察：agentscope 已经有 `_execute_tool_calls_concurrently`，通过 `asyncio.gather` 并行执行多个 tool call。一个内部创建临时 `Agent` 实例的 `spawn_subagent` 工具，天然从现有基础设施获得并行能力。

**Sub-agent 特征：**

| 维度 | 主 agent | Sub-agent |
|------|---------|-----------|
| 生命周期 | 跟随 session | 临时，task 完成即销毁 |
| 工具集 | 全量（编排 + 执行） | 仅执行 |
| System prompt | 完整 | 精简的任务执行 prompt |
| 状态 | 拥有 `AgentState` | 无 |
| Memory | 完整对话历史 | 仅任务描述 + 上下文 |
| 事件 | `metadata = {}` | `metadata = {subagent_id: "..."}` |

**工具签名：**

```python
async def spawn_subagent(
task: str,
context: str | None = None,
# 现有参数保留，向后兼容：
fork: bool = False,
background: bool = False,
timeout: int = 600,
) -> ToolResponse:
```

当 `fork=False` 且 `background=False` 时走进程内路径。向后兼容 — 现有 `fork=True` 和 `background=True` 路径不变。

**流式事件与最终返回 — 两层设计：**

Sub-agent 是一个 ReAct agent，执行过程中持续产出流式事件（thinking、tool call、文本输出）。数据消费方有两个，需要区分两层：

| 层 | 载体 | 消费方 | 用途 |
|---|---|---|---|
| 执行过程 | `EventBase`（SSE 推送） | 前端 | 实时展示 sub-agent 的 thinking、tool call、文本输出 |
| 执行结束 | `ToolResponse`（工具返回值） | 主 agent 的 LLM | 主 agent 据此决定下一步 |

**前提：扩展工具执行管道支持 `EventBase` 透传**

当前工具只能 yield `ToolChunk`/`ToolResponse`。Sub-agent 的事件（`ThinkingBlockDeltaEvent`、`ToolCallStartEvent`、`TextBlockDeltaEvent` 等）是 `EventBase` 子类——无法到达 `reply_stream()`。需要三个 agentscope 改动：

1. **`toolkit.call_tool`**（`agentscope/tool/_toolkit.py:321-324`）：遍历工具 async generator 输出时，遇到 `EventBase` 直接 passthrough yield，不累积到 `ToolResponse`：

```python
elif isinstance(res, AsyncGenerator):
async for chunk in res:
if isinstance(chunk, EventBase):
yield chunk # passthrough，不累积
else:
yield chunk
tool_response.append_chunk(chunk)
```

2. **`_execute_tool_call`**（`agentscope/agent/_agent.py:1380-1466`）：遍历 `_acting()` 时，遇到 `EventBase` 直接 yield 到 `reply_stream()`，不走 `_convert_tool_chunk_to_event`：

```python
async for chunk in self._acting(tool_call):
if isinstance(chunk, ToolResponse):
# 现有处理：save to context, yield ToolResultEndEvent
...
elif isinstance(chunk, EventBase):
# 新增：sub-agent 事件直接透传
yield chunk
else:
# 现有 ToolChunk 处理
...
```

3. **类型签名**：`_acting`、`_acting_impl`、`call_tool` 的 yield 类型从 `ToolChunk | ToolResponse` 扩展为 `ToolChunk | ToolResponse | EventBase`。

这是通用扩展——任何内部运行 agent（或其他事件生产者）的工具都受益。`spawn_subagent` 遍历 `sub_agent.reply_stream()`，在每个事件上注入 `metadata`，作为 `EventBase` yield 出去。

**流式事件（前端消费）：**

Sub-agent 产出的每个 `EventBase` 通过工具管道透传到 `reply_stream()`，带 metadata：

```python
metadata = {"subagent_id": "sa_7kX2m"}
```

前端按 `metadata` 路由：有 `subagent_id` → 展示在 side panel（如任务/DAG 节点面板）；无 → 展示在 chat 面板。应用可以追加自己的 key（如 DataPaw 追加 `graph_id` + `node_id` 做 DAG 节点路由）。

**最终返回（主 agent 消费）：**

Sub-agent 结束后，`spawn_subagent` yield 一个 `ToolChunk` 作为主 agent 的 tool result。累积的 `ToolResponse` 复用其已有字段：

```python
ToolResponse(
state=ToolResultState.SUCCESS, # 或 ERROR
content=[TextBlock(text="...")], # sub-agent 的最终文本输出
metadata={ # 结构化结果，供程序化访问
"subagent_id": "sa_7kX2m",
"files": [ # 产出的 artifact（如有）
{"name": "report.csv", "path": "/workspace/artifacts/report.csv", "mime_type": "text/csv"},
],
},
)
```

| 字段 | 内容 |
|------|------|
| `state` | `SUCCESS` sub-agent 正常完成；`ERROR` 失败或超时 |
| `content` | Sub-agent 的最终文本输出（`TextBlock`）— 分析结果、数据摘要、错误描述等。主 agent 的 LLM 读这个 |
| `metadata` | `subagent_id`（始终存在）、`files`（产出 artifact，可选）。应用可追加领域 key（如 DataPaw 追加 `node_id`）。主 agent 的工具执行层可直接读 `metadata`，无需解析 `content` 文本 |

失败示例：

```python
ToolResponse(
state=ToolResultState.ERROR,
content=[TextBlock(text="连接数据库失败：端口 5432 连接被拒绝")],
metadata={"subagent_id": "sa_7kX2m"},
)
```

**事件流全景：**

```
主 agent reply_stream()
│
├─ tool_call: spawn_subagent(task="A") ─┐
├─ tool_call: spawn_subagent(task="B") ─┤ asyncio.gather（已有基础设施）
│ │
│ sub-agent-1（ReAct 循环）： │ sub-agent-2（ReAct 循环）：
│ 流式 EventBase with │ 流式 EventBase with
│ metadata {subagent_id: "sa_1"} │ metadata {subagent_id: "sa_2"}
│ → 前端实时展示进度 │ → 前端实时展示进度
│ │
│ 最终 ToolResponse │ 最终 ToolResponse
│ → 主 agent 读 content │ → 主 agent 读 content
│ │
└─── 主 agent 继续 reasoning ────────────┘
```

**错误处理：**

- 单个 sub-agent 失败不取消其他（与 `_execute_tool_calls_concurrently` 一致）
- 失败 sub-agent 以 `ToolResponse(state=ERROR)` 返回错误文本
- 主 agent 决定重试 / 跳过 / 标记失败

### 任务清单

- [ ] **T1**: 给 `agentscope.event.EventBase` 加 `metadata: dict[str, Any]`
- [ ] **T2**: 扩展工具执行管道 — `toolkit.call_tool`、`_execute_tool_call`、`_acting` 支持 `EventBase` 与 `ToolChunk`/`ToolResponse` 并行透传
- [ ] **T3**: 实现 `spawn_subagent` 的进程内路径（`fork=False, background=False` 时），通过管道 yield sub-agent 的 `EventBase` 事件（带 metadata）
- [ ] **T4**: 文档：metadata 约定和 sub-agent 使用模式

### 落地顺序建议

1. **T1 + T2**（协议）：T1 是 `EventBase` 加一个字段。T2 扩展工具管道支持 `EventBase` 透传 — 两者是 sub-agent 事件路由的前提。
2. **T3**（进程内 sub-agent）：依赖 T1 + T2。实现 `spawn_subagent` 的进程内执行 + 事件流。
3. **T4**（文档）：全部落地后。

## Alternatives Considered

- **保留 HTTP-only `spawn_subagent`**：适合跨 agent 委托，但 sub-agent 事件对父 agent 的流不可见，无法 inline 展示并行子任务进度。
- **应用层 side-channel 事件合并**（不改上游，应用 SSE 端点自行合并）：应用的 SSE endpoint 合并 `reply_stream()` 和一个 side-channel broadcaster（sub-agent 工具往里推事件）。不需要上游改动，但：(1) 每个应用都要实现自己的合并逻辑，(2) sub-agent 事件绕过 agentscope 的事件管道（middleware/interceptor 看不到），(3) 并发事件排序在框架外更难保证。管道扩展（Proposal 2）更通用、更可组合。
- **加一个 `spawn_parallel(node_ids=[...])` 批量工具**：否决 — 违反原子性。主 agent 应通过发多个原子 `spawn_subagent` 调用来决定并行度，而非传数组。更简单、更灵活（同一 node 可以有多个 sub-agent），且与 LLM 天然发出并行 tool call 的方式一致。
- **给每种事件类型做 subclass 加 metadata**：不同应用无法组合。middleware/interceptor 无法通用地访问 metadata。
- **用 `reply_id` 做 sub-agent 路由**：不够 — `reply_id` 标识一次 reply，不标识一个 sub-agent。同一 tool-call turn 里 spawn 的多个 sub-agent 共享父 agent 的 `reply_id`。

## Additional Context

- 现有 `spawn_subagent` 实现：`src/qwenpaw/agents/tools/agent_management.py:646-745`
- agentscope `EventBase`：`agentscope/event/_event.py:53-62`
- agentscope 并行工具执行：`agentscope/agent/_agent.py:1130-1204`（`_execute_tool_calls_concurrently`，用 `asyncio.gather`）
- DataPaw sub-agent 设计文档：`datapaw/design-docs/2026-06-02-sub-agent-parallel-execution.md`
- 关联 issue：DataPaw plugin API 扩展（`datapaw-docs/2026-05-28-host-plugin-api-issue.md`，T1/T2 覆盖运行时层面的 Event metadata）

## Willing to Contribute

- [x] I am willing to open a PR for this feature (after discussion).

DataPaw 维护方将贡献实现和测试，可提供 DAG 并行执行场景的具体使用数据。