From 2748e5e4fe507360523af49df7dbf98491e46863 Mon Sep 17 00:00:00 2001 From: gxgeek-n <189542381+gxgeek-n@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:15:00 +0800 Subject: [PATCH 1/2] fix(harness): emit AgentEndEvent for sync-spawned subagents on parent cancel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AgentSpawnTool emits an AgentStartEvent unconditionally before subscribing to the child, then emits the matching AgentEndEvent from doOnTerminate. Reactor's doOnTerminate fires on onComplete/onError but not on cancel, so cancelling the parent (user "Stop", an outer timeout(), or any upstream Reactor cancel) leaves the child's event stream open: consumers saw a start with no end and kept rendering the subagent as running. Switched to doFinally so all three terminal signals close the stream. The same file already uses doFinally for the timeout-promotion path (and its comment explicitly reasons about doFinally(CANCEL)), so this was an oversight rather than intent. Verified Reactor semantics directly before changing anything: complete doOnTerminate=1 doFinally=1 error doOnTerminate=1 doFinally=1 cancel doOnTerminate=0 doFinally=1 Adds AgentSpawnToolCancelEndEventTest with two cases: - parent cancel must still emit AgentEndEvent — fails without this change (expected 1, got 0) - normal completion emits exactly one start and one end — passes both ways, so the fix does not introduce duplicate events on the happy path Full agentscope-harness suite: 693 tests, 0 failures. Scope note: distinct from #2408/#2412, which fix whether the child execution actually stops (interruptAgent being a no-op for HarnessAgent). Those touch interruptAgent and the child RuntimeContext; this touches only the event pairing. Even with the child correctly interrupted, the event stream still has to close. --- .../harness/agent/tool/AgentSpawnTool.java | 7 +- .../AgentSpawnToolCancelEndEventTest.java | 187 ++++++++++++++++++ 2 files changed, 192 insertions(+), 2 deletions(-) create mode 100644 agentscope-harness/src/test/java/io/agentscope/harness/agent/tool/AgentSpawnToolCancelEndEventTest.java diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/tool/AgentSpawnTool.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/tool/AgentSpawnTool.java index 9949d6a3a7..6841938b2f 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/tool/AgentSpawnTool.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/tool/AgentSpawnTool.java @@ -671,8 +671,11 @@ private Mono execLocalSync( c.put( AgentEventEmitter.FORWARDING_CONTEXT_KEY, taggedEmitter)) - .doOnTerminate( - () -> + // doFinally, not doOnTerminate: the latter skips cancel, so a + // parent cancel would leave the AgentStartEvent above unmatched + // and consumers would render this subagent as running forever. + .doFinally( + signal -> parentEmitter.emit( new AgentEndEvent(null) .withSource(sourcePath))); diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/tool/AgentSpawnToolCancelEndEventTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/tool/AgentSpawnToolCancelEndEventTest.java new file mode 100644 index 0000000000..a9f54067f2 --- /dev/null +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/tool/AgentSpawnToolCancelEndEventTest.java @@ -0,0 +1,187 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.harness.agent.tool; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; + +import io.agentscope.core.ReActAgent; +import io.agentscope.core.agent.RuntimeContext; +import io.agentscope.core.event.AgentEndEvent; +import io.agentscope.core.event.AgentEvent; +import io.agentscope.core.event.AgentEventEmitter; +import io.agentscope.core.event.AgentStartEvent; +import io.agentscope.core.message.Msg; +import io.agentscope.harness.agent.HarnessAgent; +import io.agentscope.harness.agent.middleware.SubagentEntry; +import io.agentscope.harness.agent.subagent.DefaultAgentManager; +import io.agentscope.harness.agent.subagent.task.BackgroundTask; +import io.agentscope.harness.agent.subagent.task.TaskRepository; +import io.agentscope.harness.agent.subagent.task.TaskRunSpec; +import io.agentscope.harness.agent.subagent.task.TaskStatus; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.mockito.Mockito; +import reactor.core.Disposable; +import reactor.core.publisher.Mono; + +/** + * {@code AgentStartEvent} / {@code AgentEndEvent} pairing for sync-spawned subagents. + * + *

{@code AgentSpawnTool} emits an {@link AgentStartEvent} unconditionally before subscribing to + * the child, then emits the matching {@link AgentEndEvent} from a terminate callback. Reactor's + * {@code doOnTerminate} only fires on {@code onComplete} / {@code onError} — not on cancel — + * so cancelling the parent (user "Stop", an outer {@code timeout()}, or any upstream Reactor cancel) + * leaves the child's event stream open forever: the consumer saw a start with no end and keeps + * rendering the subagent as running. + * + *

Note this is distinct from whether the child execution itself stops (see #2408 / #2412, which + * fix {@code interruptAgent} being a no-op for {@code HarnessAgent}). Even once the child is + * correctly interrupted, the emitted event stream still has to close. + */ +@DisplayName("AgentSpawnTool parent-cancel: subagent event stream must close") +class AgentSpawnToolCancelEndEventTest { + + /** Collects every event the tool emits into the parent's stream. */ + private static final class RecordingEmitter implements AgentEventEmitter { + private final List events = new CopyOnWriteArrayList<>(); + + @Override + public void emit(AgentEvent event) { + events.add(event); + } + + long count(Class type) { + return events.stream().filter(type::isInstance).count(); + } + } + + @Test + @DisplayName("parent cancel still emits AgentEndEvent for the spawned child") + @Timeout(value = 10, unit = TimeUnit.SECONDS) + void parentCancel_emitsAgentEndEvent() throws Exception { + CountDownLatch childStarted = new CountDownLatch(1); + + ReActAgent delegate = Mockito.mock(ReActAgent.class); + HarnessAgent harness = Mockito.mock(HarnessAgent.class); + Mockito.when(harness.getDelegate()).thenReturn(delegate); + // Child never finishes on its own, so the only way this Mono terminates is cancel. + Mockito.when(harness.call(any(Msg.class), any(RuntimeContext.class))) + .thenReturn(Mono.never().doOnSubscribe(ignored -> childStarted.countDown())); + + DefaultAgentManager manager = + new DefaultAgentManager( + List.of(new SubagentEntry("harness_agent", "Harness child", rc -> harness)), + null); + AgentSpawnTool tool = new AgentSpawnTool(manager, new NoopTaskRepository(), 0); + RuntimeContext parentCtx = + RuntimeContext.builder().sessionId("parent-session").userId("parent-user").build(); + + RecordingEmitter emitter = new RecordingEmitter(); + + Disposable subscription = + tool.agentSpawn(parentCtx, null, "harness_agent", "work", null, 30, null) + .contextWrite(ctx -> ctx.put(AgentEventEmitter.CONTEXT_KEY, emitter)) + .subscribe(); + + assertTrue(childStarted.await(5, TimeUnit.SECONDS), "child should have started"); + assertEquals( + 1, + emitter.count(AgentStartEvent.class), + "a start event should have been emitted for the spawned child"); + + subscription.dispose(); + + // Give the cancel signal a moment to run the terminate/finally callbacks. + Thread.sleep(300); + + assertEquals( + 1, + emitter.count(AgentEndEvent.class), + "parent cancel must still close the child's event stream — an AgentStartEvent" + + " without a matching AgentEndEvent leaves consumers rendering the" + + " subagent as running forever (doOnTerminate does not fire on cancel)"); + } + + @Test + @DisplayName("normal completion emits exactly one start and one end") + @Timeout(value = 10, unit = TimeUnit.SECONDS) + void normalCompletion_emitsPairedEvents() throws Exception { + ReActAgent delegate = Mockito.mock(ReActAgent.class); + HarnessAgent harness = Mockito.mock(HarnessAgent.class); + Mockito.when(harness.getDelegate()).thenReturn(delegate); + Mockito.when(harness.call(any(Msg.class), any(RuntimeContext.class))) + .thenReturn(Mono.just(Msg.builder().name("child").textContent("done").build())); + + DefaultAgentManager manager = + new DefaultAgentManager( + List.of(new SubagentEntry("harness_agent", "Harness child", rc -> harness)), + null); + AgentSpawnTool tool = new AgentSpawnTool(manager, new NoopTaskRepository(), 0); + RuntimeContext parentCtx = + RuntimeContext.builder().sessionId("parent-session").userId("parent-user").build(); + + RecordingEmitter emitter = new RecordingEmitter(); + + tool.agentSpawn(parentCtx, null, "harness_agent", "work", null, 30, null) + .contextWrite(ctx -> ctx.put(AgentEventEmitter.CONTEXT_KEY, emitter)) + .block(); + + assertEquals(1, emitter.count(AgentStartEvent.class), "expected one start event"); + assertEquals(1, emitter.count(AgentEndEvent.class), "expected one end event"); + } + + private static final class NoopTaskRepository implements TaskRepository { + @Override + public BackgroundTask getTask(RuntimeContext rc, String sessionId, String taskId) { + return null; + } + + @Override + public BackgroundTask putTask( + RuntimeContext rc, + String taskId, + String subAgentId, + String sessionId, + TaskRunSpec spec) { + return null; + } + + @Override + public void removeTask(RuntimeContext rc, String sessionId, String taskId) {} + + @Override + public void clear() {} + + @Override + public Collection listTasks( + RuntimeContext rc, String sessionId, TaskStatus filter) { + return List.of(); + } + + @Override + public boolean cancelTask(RuntimeContext rc, String sessionId, String taskId) { + return false; + } + } +} From 4e58a1a185a34499f977db43488489512537fce7 Mon Sep 17 00:00:00 2001 From: gxgeek-n <189542381+gxgeek-n@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:27:15 +0800 Subject: [PATCH 2/2] test(harness): replace fixed sleep with a bounded wait for the end event Review feedback: Thread.sleep(300) after dispose() makes the test slower than needed and can still be flaky if a slow CI machine delays the cancel callbacks. Now polls until AgentEndEvent is observed, with a 5 s ceiling and a 20 ms interval, so it returns as soon as the event lands. --- .../agent/tool/AgentSpawnToolCancelEndEventTest.java | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/tool/AgentSpawnToolCancelEndEventTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/tool/AgentSpawnToolCancelEndEventTest.java index a9f54067f2..4773ee6b5f 100644 --- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/tool/AgentSpawnToolCancelEndEventTest.java +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/tool/AgentSpawnToolCancelEndEventTest.java @@ -112,8 +112,12 @@ void parentCancel_emitsAgentEndEvent() throws Exception { subscription.dispose(); - // Give the cancel signal a moment to run the terminate/finally callbacks. - Thread.sleep(300); + // Bounded wait rather than a fixed sleep: returns as soon as the end event lands, and does + // not turn flaky if a slow CI machine delays the cancel callbacks. + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (emitter.count(AgentEndEvent.class) == 0 && System.nanoTime() < deadline) { + Thread.sleep(20); + } assertEquals( 1,