fix(core): restore event emitter for detached tool calls

[wlmwang]

fix: #2482

AgentScope-Java Version

2.0.1-SNAPSHOT

Description

修改概述

修复 HarnessAgent#streamEvents() 中，本地同步子 Agent 通过 agent_spawn 启动后，子 Agent 的 AgentEvent 无法转发到父 Agent 事件流的问题。

问题原因

streamEvents() 创建事件流时会把当前订阅对应的 FluxSink<AgentEvent> 放入 Reactor Context，并在 ReActAgent#doCall(...) 中绑定到当前 CallExecution.eventSink。

acting 阶段执行工具调用时，工具执行链路经过内部 executeToolCalls(...) 订阅。该链路无法继续读取原始 Reactor Context 中的 AgentEventEmitter.CONTEXT_KEY。因此 AgentSpawnTool#execLocalSync(...) 调用 AgentEventEmitter.fromContext(ctxView) 时拿不到父事件 emitter，子 Agent 事件无法推送到父事件流。

修复方案

在 executeToolCalls(...) 的 contextWrite(...) 边界中，先通过 ctx.putAll(parentCtx) 合并调用侧的 Reactor ContextView。由于 acting 阶段工具调用使用内部订阅执行，原始 streamEvents() 订阅中的 event emitter 没有继续出现在工具执行链路里；因此当合并后的 context 中不存在 SubagentEventBus.CONTEXT_KEY 和 AgentEventEmitter.CONTEXT_KEY 时，会从当前 CallExecution 已绑定的 eventSink 或 externalEventEmitter 恢复 AgentEventEmitter.CONTEXT_KEY：

• 优先使用当前 streamEvents() 订阅绑定的 eventSink；
• 当当前调用本身是被父 Agent 转发调用时，回退使用 externalEventEmitter；
• 保留已有 SubagentEventBus.CONTEXT_KEY 和 AgentEventEmitter.CONTEXT_KEY，避免改变 deprecated stream() 或显式转发链路的行为。

这样 AgentSpawnTool#execLocalSync(...) 可以在 acting 工具调用链路中重新获取父级 emitter，并把子 Agent 事件带 source 转发到父事件流。

测试覆盖

• 新增 ReActAgentNewLoopE2ETest 覆盖 acting context 丢失事件 key 时，executeToolCalls(...) 会从当前 eventSink 恢复 AgentEventEmitter.CONTEXT_KEY。
• 覆盖 streamEvents() 首尾仍为父 Agent AGENT_START / AGENT_END。
• 保留并验证 HarnessAgentSubagentStreamEventsTest 中子 Agent 事件会带 source 转发到父事件流。
• 在 harness 子 Agent 流式事件测试中关闭 memory hooks，避免与事件转发无关的 session JSONL 写入影响 @TempDir 清理。

[gemini-code-assist]

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

[codecov]

Codecov Report

❌ Patch coverage is 66.66667% with 4 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
...e/src/main/java/io/agentscope/core/ReActAgent.java	66.66%	3 Missing and 1 partial ⚠️

📢 Thoughts on this report? Let us know!

[oss-maintainer]

Review — Approved ✅

Fix: Restore AgentEventEmitter in Reactor Context when executeToolCalls(...) creates an internal subscription that loses the original streaming context.

Analysis

The root cause is clear: streamEvents() puts the AgentEventEmitter into Reactor Context at subscription time, but executeToolCalls(...) creates a new internal subscription. The contextWrite(ctx -> ctx.putAll(parentCtx)) merges the parent context, but if the parent context itself lost the emitter key (e.g., through a middleware boundary or context isolation), the child agent spawned in AgentSpawnTool#execLocalSync(...) can't find the parent emitter and falls back to non-streaming execution.

The fix adds a fallback in the contextWrite lambda:

1. Merge parent context as before
2. If both SubagentEventBus.CONTEXT_KEY and AgentEventEmitter.CONTEXT_KEY are missing:
   • Prefer eventSink (local streaming subscription) → wrap as AgentEventEmitter
   • Fall back to externalEventEmitter (parent agent forwarding)
3. If either key already exists, preserve it (no override)

This is correct because:

• eventSink is bound to the current CallExecution during streamEvents() setup, so it's always available when streaming
• externalEventEmitter handles the case where this agent was itself called by a parent agent's forwarding
• The guard !merged.hasKey(...) ensures backward compatibility with stream() (deprecated) and explicit event forwarding paths

Test Quality

The test uses a StripToolEventContextMiddleware to simulate the exact failure mode — stripping AgentEventEmitter.CONTEXT_KEY before tool execution. This is a clever and precise test design that:

• Verifies the fix restores the emitter from eventSink
• Confirms AGENT_START / AGENT_END events are still emitted
• Validates exactly 1 ToolResultEndEvent (no duplicates)
• Tests both the "eventSink available" and "externalEventEmitter fallback" paths

Minor Notes

• The lambda cast (AgentEventEmitter) eventSink::next works because AgentEventEmitter is a functional interface. Consider adding a static factory method like AgentEventEmitter.fromSink(FluxSink) for clarity if this pattern appears elsewhere.
• Thread safety: FluxSink.next() is thread-safe per Reactor docs, so concurrent tool calls won't corrupt the event stream.

Verdict: Clean, minimal, well-tested fix for a real Context propagation gap. LGTM.