diff --git a/mateclaw-server/src/main/java/vip/mate/task/AsyncTaskService.java b/mateclaw-server/src/main/java/vip/mate/task/AsyncTaskService.java index 60e383fe..c179b394 100644 --- a/mateclaw-server/src/main/java/vip/mate/task/AsyncTaskService.java +++ b/mateclaw-server/src/main/java/vip/mate/task/AsyncTaskService.java @@ -508,6 +508,12 @@ public class AsyncTaskService implements ApplicationRunner { data.put("taskId", task.getTaskId()); data.put("taskType", task.getTaskType()); data.put("success", success); + data.put("status", Objects.toString(task.getStatus(), success ? "succeeded" : "failed")); + if (task.getProgress() != null) data.put("progress", task.getProgress()); + if (task.getCreateTime() != null && task.getUpdateTime() != null) { + data.put("durationMs", Math.max(0L, + java.time.Duration.between(task.getCreateTime(), task.getUpdateTime()).toMillis())); + } if (extraData != null) data.putAll(extraData); if (errorMessage != null) data.put("errorMessage", errorMessage); streamTracker.broadcastObject(task.getConversationId(), eventName, data); diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java index 9136b428..96dcd7b4 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java @@ -176,6 +176,18 @@ public class DelegateAgentTool { /** Polling interval inside {@code block=true} wait. */ private static final long TASK_OUTPUT_POLL_INTERVAL_MS = 500L; + /** Omitted async delegation timeouts are bounded instead of running forever. */ + private static final int ASYNC_DELEGATION_DEFAULT_TIMEOUT_S = 3600; + + /** Keep an accidentally huge model-supplied timeout from creating an effectively immortal task. */ + private static final int ASYNC_DELEGATION_MAX_TIMEOUT_S = 86_400; + + /** Brief grace period for graph/tool cancellation hooks to finish cleanup. */ + private static final int ASYNC_DELEGATION_CANCEL_GRACE_S = 2; + + @Value("${mateclaw.delegation.async-timeout-seconds:3600}") + private int asyncDelegationTimeoutSeconds; + /** * Operator-supplied deny-list extension. Configured via * {@code mateclaw.delegation.child-denied-tools} as a comma-separated @@ -791,6 +803,8 @@ public class DelegateAgentTool { @ToolParam(description = "Task description with complete context information") String task, @ToolParam(description = "Optional short label (≤ 32 chars) for human tracking on the UI badge", required = false) String label, + @ToolParam(description = "Optional execution budget in seconds. Default 3600, max 86400. Timeout stops the child session and persists a failed task result.", + required = false) Integer timeoutSeconds, @Nullable ToolContext ctx) { if (agentName == null || agentName.isBlank()) { @@ -799,6 +813,12 @@ public class DelegateAgentTool { if (task == null || task.isBlank()) { return errorJson("task 不能为空"); } + int effectiveTimeoutSeconds; + try { + effectiveTimeoutSeconds = resolveAsyncTimeoutSeconds(timeoutSeconds); + } catch (IllegalArgumentException e) { + return errorJson(e.getMessage()); + } String safeLabel = label == null ? "" : (label.length() > ASYNC_LABEL_MAX_CHARS ? label.substring(0, ASYNC_LABEL_MAX_CHARS) : label); @@ -856,6 +876,7 @@ public class DelegateAgentTool { payload.put("depth", childDepth); payload.put("task", truncate(task, ASYNC_TASK_REQUEST_MAX_CHARS)); payload.put("label", safeLabel); + payload.put("timeoutSeconds", effectiveTimeoutSeconds); requestJson = objectMapper.writeValueAsString(payload); } catch (Exception e) { subagentRegistry.unregister(subagentId); @@ -875,9 +896,10 @@ public class DelegateAgentTool { // Detached async child: its usage belongs to the later // task_output retrieval, not the spawning turn, so do not // roll it into the parent's _usage_final. - ChildResult childResult = runSingleChild(0, target, task, - parentConversationId, childConversationId, parentOrigin, - rootConvAsync, subagentId, childDepth, false); + ChildResult childResult = runDetachedChildWithTimeout( + target, task, parentConversationId, childConversationId, + parentOrigin, rootConvAsync, subagentId, childDepth, + effectiveTimeoutSeconds); return childResult.toToolResponse(target.getName()); } finally { subagentRegistry.get(subagentId).ifPresent(rec -> { @@ -917,6 +939,7 @@ public class DelegateAgentTool { result.put("child_conversation_id", childConversationId); result.put("agent_name", target.getName()); result.put("status", "running"); + result.put("timeout_seconds", effectiveTimeoutSeconds); result.put("hint", "Call task_output(task_id) in a later turn to retrieve the result."); if (!safeLabel.isEmpty()) { result.put("label", safeLabel); @@ -1074,6 +1097,78 @@ public class DelegateAgentTool { return Duration.between(entity.getCreateTime(), entity.getUpdateTime()).toMillis(); } + int resolveAsyncTimeoutSeconds(Integer requested) { + int configuredDefault = asyncDelegationTimeoutSeconds > 0 + ? asyncDelegationTimeoutSeconds + : ASYNC_DELEGATION_DEFAULT_TIMEOUT_S; + int resolved = requested != null ? requested : configuredDefault; + if (resolved <= 0 || resolved > ASYNC_DELEGATION_MAX_TIMEOUT_S) { + throw new IllegalArgumentException("timeoutSeconds must be between 1 and " + + ASYNC_DELEGATION_MAX_TIMEOUT_S); + } + return resolved; + } + + /** + * Execute a detached child with a real wall-clock bound. Cancelling only the + * {@link CompletableFuture} is insufficient because the graph may already be + * blocked in an LLM or tool call; requestStop gives the child runtime a + * cooperative stop signal at its next checkpoint as well. + */ + private ChildResult runDetachedChildWithTimeout( + AgentEntity target, String task, String parentConversationId, + String childConversationId, ChatOrigin parentOrigin, + String rootConversationId, String subagentId, int childDepth, + int timeoutSeconds) throws Exception { + CountDownLatch childFinished = new CountDownLatch(1); + Future future = DELEGATION_EXECUTOR.submit( + () -> { + try { + return runSingleChild(0, target, task, parentConversationId, + childConversationId, parentOrigin, rootConversationId, + subagentId, childDepth, false); + } finally { + childFinished.countDown(); + } + }); + try { + return future.get(timeoutSeconds, TimeUnit.SECONDS); + } catch (TimeoutException e) { + streamTracker.requestStop(childConversationId); + subagentRegistry.get(subagentId).ifPresent(record -> record.status().set("timeout")); + future.cancel(true); + awaitDetachedChildCleanup(childFinished, childConversationId); + log.warn("Async delegation timed out: childConv={}, agent={}, timeout={}s", + childConversationId, target.getName(), timeoutSeconds); + throw new TimeoutException("Async delegation timed out after " + timeoutSeconds + " seconds"); + } catch (InterruptedException e) { + streamTracker.requestStop(childConversationId); + subagentRegistry.get(subagentId).ifPresent(record -> record.status().set("interrupted")); + future.cancel(true); + awaitDetachedChildCleanup(childFinished, childConversationId); + Thread.currentThread().interrupt(); + throw e; + } catch (ExecutionException e) { + Throwable cause = e.getCause(); + if (cause instanceof Exception exception) throw exception; + throw new IllegalStateException("Async delegation failed", cause); + } + } + + private void awaitDetachedChildCleanup(CountDownLatch childFinished, String childConversationId) { + boolean interrupted = false; + try { + if (!childFinished.await(ASYNC_DELEGATION_CANCEL_GRACE_S, TimeUnit.SECONDS)) { + log.warn("Async delegation cancellation grace expired: childConv={}, grace={}s", + childConversationId, ASYNC_DELEGATION_CANCEL_GRACE_S); + } + } catch (InterruptedException e) { + interrupted = true; + } finally { + if (interrupted) Thread.currentThread().interrupt(); + } + } + // ==================== Child agent execution (shared by single and parallel paths) ==================== /** @@ -1089,8 +1184,12 @@ public class DelegateAgentTool { ChatOrigin parentOrigin, String rootConversationId, String subagentId, int childDepth, boolean accumulateToParent) { - boolean relayChildEvents = parentConversationId != null && streamTracker.isRunning(parentConversationId); - if (relayChildEvents) { + // Track every child run, including detached work that starts after the + // parent stream has already completed. Without its own RunState a later + // timeout can interrupt the wrapper Future but requestStop cannot reach + // graph checkpoints or registered tool-cancellation hooks. + boolean trackChildRun = childConversationId != null && !childConversationId.isBlank(); + if (trackChildRun) { streamTracker.register(childConversationId); streamTracker.incrementFlux(childConversationId); } @@ -1128,11 +1227,17 @@ public class DelegateAgentTool { return ChildResult.ofSuccess(taskIndex, target.getName(), rawResult, durationMs, MAX_RESULT_LENGTH, chatResult.promptTokens(), chatResult.completionTokens()); } catch (Exception e) { + if (e instanceof InterruptedException || e instanceof CancellationException) { + if (e instanceof InterruptedException) Thread.currentThread().interrupt(); + log.info("Child agent interrupted: taskIndex={}, agent={}, childConv={}", + taskIndex, target.getName(), childConversationId); + return ChildResult.ofCancelled(taskIndex, target.getName()); + } log.error("Child agent failed: taskIndex={}, agent={}, error={}", taskIndex, target.getName(), e.getMessage()); return ChildResult.ofError(taskIndex, target.getName(), e.getMessage()); } finally { - if (relayChildEvents) { + if (trackChildRun) { streamTracker.complete(childConversationId); } DelegationContext.exit(); diff --git a/mateclaw-server/src/main/resources/application.yml b/mateclaw-server/src/main/resources/application.yml index ea168761..4a3a6690 100644 --- a/mateclaw-server/src/main/resources/application.yml +++ b/mateclaw-server/src/main/resources/application.yml @@ -402,6 +402,10 @@ mateclaw: # models (Kimi / GLM / MiniMax) routinely take 90–290 s per LLM turn # when the child must produce multi-section structured output. parallel-timeout-seconds: 300 + # Default wall-clock budget for detached delegateAsync children. A caller + # may request a different positive timeout up to 86400 seconds; keeping a + # default bound prevents abandoned background children from running forever. + async-timeout-seconds: 3600 # MateClaw Agent 配置 mate: diff --git a/mateclaw-server/src/main/resources/docs/en/agents.md b/mateclaw-server/src/main/resources/docs/en/agents.md index 6a02f916..699b260f 100644 --- a/mateclaw-server/src/main/resources/docs/en/agents.md +++ b/mateclaw-server/src/main/resources/docs/en/agents.md @@ -121,7 +121,7 @@ Three delegation tools, one per cadence: - **`delegateToAgent`** — synchronous. Hand a sub-task to a specific employee, wait for it to finish, and return only after the child's final result. Optional `inheritParentContext` carries the parent conversation's recent context to the child, so you don't have to re-explain the background. - **`delegateParallel`** — fan out. Delegate to several children at once; each runs in its own isolated session and the results are collected together. -- **`delegateAsync`** — background. Returns a `task_id` immediately while the child runs in the background; fetch the result later with **`taskOutput`**. `taskOutput` has an **attribution gate** — only the **same conversation + the same user** that spawned the task can read its result, preventing cross-conversation / cross-user leakage. +- **`delegateAsync`** — background. Returns a `task_id` immediately while the child runs in the background; fetch the result later with **`taskOutput`**. Background runs have a bounded execution budget (default 3600 seconds, configurable with `mateclaw.delegation.async-timeout-seconds`, or per call with `timeoutSeconds`, max 86400). On timeout MateClaw stops the child session and persists a failed result instead of leaving an orphan run. `taskOutput` has an **attribution gate** — only the **same conversation + the same user** that spawned the task can read its result, preventing cross-conversation / cross-user leakage. Children deny a default set of tools so the tree can't run away: diff --git a/mateclaw-server/src/main/resources/docs/zh/agents.md b/mateclaw-server/src/main/resources/docs/zh/agents.md index 21168222..17752614 100644 --- a/mateclaw-server/src/main/resources/docs/zh/agents.md +++ b/mateclaw-server/src/main/resources/docs/zh/agents.md @@ -121,7 +121,7 @@ head: - **`delegateToAgent`** —— 同步委派。把一个子任务交给指定员工，等它跑完、拿到最终结果再返回。可选 `inheritParentContext`：把父会话最近的上下文一起带给子员工，省去重复交代背景。 - **`delegateParallel`** —— 扇出委派。同时派给多个子员工，各自在隔离会话里跑，结果统一收集回来。 -- **`delegateAsync`** —— 后台委派。立刻返回一个 `task_id`，子员工在后台跑；之后用 **`taskOutput`** 取结果。`taskOutput` 带**归属闸门**——只有最初发起委派的**同一个会话 + 同一个用户**才能读到结果，防止跨会话/跨用户泄露。 +- **`delegateAsync`** —— 后台委派。立刻返回一个 `task_id`，子员工在后台跑；之后用 **`taskOutput`** 取结果。后台任务默认有 3600 秒执行预算，可通过 `mateclaw.delegation.async-timeout-seconds` 配置，或在单次调用中用 `timeoutSeconds` 调整（最大 86400 秒）；超时后 MateClaw 会停止子会话并持久化失败结果，避免遗留孤儿任务。`taskOutput` 带**归属闸门**——只有最初发起委派的**同一个会话 + 同一个用户**才能读到结果，防止跨会话/跨用户泄露。 子员工默认被拒绝一组工具，保证树不失控： diff --git a/mateclaw-server/src/test/java/vip/mate/task/AsyncTaskServiceOneShotTest.java b/mateclaw-server/src/test/java/vip/mate/task/AsyncTaskServiceOneShotTest.java index 4bd8f853..4460d91d 100644 --- a/mateclaw-server/src/test/java/vip/mate/task/AsyncTaskServiceOneShotTest.java +++ b/mateclaw-server/src/test/java/vip/mate/task/AsyncTaskServiceOneShotTest.java @@ -10,6 +10,7 @@ import vip.mate.task.model.AsyncTaskEntity; import vip.mate.task.repository.AsyncTaskMapper; import vip.mate.workspace.conversation.event.ConversationDeletedEvent; +import java.time.LocalDateTime; import java.util.UUID; import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentHashMap; @@ -233,6 +234,34 @@ class AsyncTaskServiceOneShotTest { eq("conv-broadcast-fail"), eq("async_task_completed"), any()); } + @Test + @DisplayName("Completion event exposes terminal status, progress, and wall-clock duration") + @SuppressWarnings("unchecked") + void completionEventCarriesObservabilityFields() { + AsyncTaskEntity entity = new AsyncTaskEntity(); + entity.setTaskId("tid-observe"); + entity.setTaskType("agent_delegate"); + entity.setConversationId("conv-observe"); + entity.setStatus("failed"); + entity.setProgress(37); + entity.setCreateTime(LocalDateTime.of(2026, 8, 30, 12, 0, 0)); + entity.setUpdateTime(LocalDateTime.of(2026, 8, 30, 12, 0, 2)); + + service.broadcastTaskEventWithData(entity, "async_task_completed", false, + java.util.Map.of("reason", "timeout"), "timed out"); + + org.mockito.ArgumentCaptor> payload = + org.mockito.ArgumentCaptor.forClass(java.util.Map.class); + verify(tracker).broadcastObject(eq("conv-observe"), + eq("async_task_completed"), payload.capture()); + assertThat(payload.getValue()) + .containsEntry("status", "failed") + .containsEntry("progress", 37) + .containsEntry("durationMs", 2_000L) + .containsEntry("reason", "timeout") + .containsEntry("errorMessage", "timed out"); + } + @Test @DisplayName("schedule/put race stress: 200 zero-cost tasks all succeed and bookkeeping drains") void scheduleAndPutRaceStress() throws Exception { diff --git a/mateclaw-server/src/test/java/vip/mate/tool/builtin/DelegateAsyncToolTest.java b/mateclaw-server/src/test/java/vip/mate/tool/builtin/DelegateAsyncToolTest.java index acbe68b6..5e62544a 100644 --- a/mateclaw-server/src/test/java/vip/mate/tool/builtin/DelegateAsyncToolTest.java +++ b/mateclaw-server/src/test/java/vip/mate/tool/builtin/DelegateAsyncToolTest.java @@ -27,8 +27,12 @@ import java.time.LocalDateTime; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyLong; @@ -104,13 +108,14 @@ class DelegateAsyncToolTest { .thenReturn(entity); when(streamTracker.isRunning("parent-conv-1")).thenReturn(true); - String result = tool.delegateAsync("Researcher", "Go research things", "label-x", makeCtx("user-1", "parent-conv-1")); + String result = tool.delegateAsync("Researcher", "Go research things", "label-x", null, makeCtx("user-1", "parent-conv-1")); Map parsed = objectMapper.readValue(result, new TypeReference<>() {}); assertThat(parsed).containsEntry("task_id", "tid-123") .containsEntry("status", "running") .containsEntry("agent_name", "Researcher") - .containsEntry("label", "label-x"); + .containsEntry("label", "label-x") + .containsEntry("timeout_seconds", 3600); assertThat((String) parsed.get("child_conversation_id")).startsWith("child-"); assertThat((String) parsed.get("hint")).contains("task_output"); @@ -132,7 +137,7 @@ class DelegateAsyncToolTest { when(asyncTaskService.submitOneShot(anyString(), anyString(), any(), anyString(), anyString(), any())) .thenReturn(entity); - tool.delegateAsync("Researcher", "task body", "myLabel", makeCtx("user-1", "parent-conv-1")); + tool.delegateAsync("Researcher", "task body", "myLabel", 7200, makeCtx("user-1", "parent-conv-1")); org.mockito.ArgumentCaptor jsonCaptor = org.mockito.ArgumentCaptor.forClass(String.class); verify(asyncTaskService).submitOneShot( @@ -142,6 +147,7 @@ class DelegateAsyncToolTest { assertThat(payload).containsEntry("parentConversationId", "parent-conv-1") .containsEntry("label", "myLabel") .containsEntry("task", "task body") + .containsEntry("timeoutSeconds", 7200) // Durable async identity — task_output's route-B authorization reads // these persisted fields (the registry is process-local), so lock them. .containsEntry("rootConversationId", "parent-conv-1") @@ -164,7 +170,7 @@ class DelegateAsyncToolTest { when(asyncTaskService.submitOneShot(anyString(), anyString(), any(), anyString(), anyString(), any())) .thenThrow(new IllegalStateException("已达到最大并行任务数（3），请等待现有任务完成")); - String result = tool.delegateAsync("Researcher", "task", null, makeCtx("user-1", "parent-conv-1")); + String result = tool.delegateAsync("Researcher", "task", null, null, makeCtx("user-1", "parent-conv-1")); Map parsed = objectMapper.readValue(result, new TypeReference<>() {}); assertThat(parsed).containsEntry("error", true); @@ -180,8 +186,8 @@ class DelegateAsyncToolTest { @Test @DisplayName("Missing agentName / task → error JSON without touching downstream services") void delegateAsyncMissingArgs() throws Exception { - String r1 = tool.delegateAsync("", "task", null, makeCtx("user-1", "parent-conv-1")); - String r2 = tool.delegateAsync("X", " ", null, makeCtx("user-1", "parent-conv-1")); + String r1 = tool.delegateAsync("", "task", null, null, makeCtx("user-1", "parent-conv-1")); + String r2 = tool.delegateAsync("X", " ", null, null, makeCtx("user-1", "parent-conv-1")); for (String r : new String[]{r1, r2}) { Map parsed = objectMapper.readValue(r, new TypeReference<>() {}); assertThat(parsed).containsEntry("error", true); @@ -195,7 +201,7 @@ class DelegateAsyncToolTest { @DisplayName("Agent not found → error JSON") void delegateAsyncAgentNotFound() throws Exception { when(agentMapper.selectOne(any())).thenReturn(null); - String result = tool.delegateAsync("Ghost", "task", null, makeCtx("user-1", "parent-conv-1")); + String result = tool.delegateAsync("Ghost", "task", null, null, makeCtx("user-1", "parent-conv-1")); Map parsed = objectMapper.readValue(result, new TypeReference<>() {}); assertThat(parsed).containsEntry("error", true); assertThat((String) parsed.get("message")).contains("Ghost"); @@ -209,7 +215,7 @@ class DelegateAsyncToolTest { when(agentMapper.selectOne(any())).thenReturn(target); when(subagentRegistry.isSpawnPaused("parent-conv-1")).thenReturn(true); - String result = tool.delegateAsync("Researcher", "task", null, makeCtx("user-1", "parent-conv-1")); + String result = tool.delegateAsync("Researcher", "task", null, null, makeCtx("user-1", "parent-conv-1")); Map parsed = objectMapper.readValue(result, new TypeReference<>() {}); assertThat(parsed).containsEntry("error", true); assertThat((String) parsed.get("message")).contains("paused"); @@ -225,13 +231,74 @@ class DelegateAsyncToolTest { for (int i = 0; i < 3; i++) { DelegationContext.enter("parent-conv-1", java.util.Set.of()); } - String result = tool.delegateAsync("Researcher", "task", null, makeCtx("user-1", "parent-conv-1")); + String result = tool.delegateAsync("Researcher", "task", null, null, makeCtx("user-1", "parent-conv-1")); Map parsed = objectMapper.readValue(result, new TypeReference<>() {}); assertThat(parsed).containsEntry("error", true); assertThat((String) parsed.get("message")).contains("depth"); verify(asyncTaskService, never()).submitOneShot(any(), any(), any(), any(), any(), any()); } + @Test + @DisplayName("delegateAsync rejects non-positive and over-cap execution budgets") + void delegateAsyncRejectsInvalidTimeout() throws Exception { + String zero = tool.delegateAsync("Researcher", "task", null, 0, + makeCtx("user-1", "parent-conv-1")); + String overCap = tool.delegateAsync("Researcher", "task", null, 86_401, + makeCtx("user-1", "parent-conv-1")); + + for (String result : new String[]{zero, overCap}) { + Map parsed = objectMapper.readValue(result, new TypeReference<>() {}); + assertThat(parsed).containsEntry("error", true); + assertThat((String) parsed.get("message")).contains("timeoutSeconds"); + } + verify(asyncTaskService, never()).submitOneShot(any(), any(), any(), any(), any(), any()); + } + + @Test + @DisplayName("delegateAsync timeout stops the child conversation and fails the persisted worker") + void delegateAsyncTimeoutPropagatesStop() throws Exception { + AgentEntity target = makeAgent(10L, "Researcher"); + when(agentMapper.selectOne(any())).thenReturn(target); + when(subagentRegistry.register(anyString(), anyString(), anyLong(), anyString(), any(), + any(), anyInt(), anyString())) + .thenReturn("sa-timeout"); + when(subagentRegistry.get("sa-timeout")).thenReturn(java.util.Optional.empty()); + + org.mockito.ArgumentCaptor> workCaptor = + org.mockito.ArgumentCaptor.forClass(Callable.class); + org.mockito.ArgumentCaptor requestCaptor = + org.mockito.ArgumentCaptor.forClass(String.class); + AsyncTaskEntity entity = new AsyncTaskEntity(); + entity.setTaskId("tid-timeout"); + when(asyncTaskService.submitOneShot( + eq("agent_delegate"), eq("parent-conv-1"), any(), requestCaptor.capture(), + eq("user-1"), workCaptor.capture())) + .thenReturn(entity); + + CountDownLatch childStarted = new CountDownLatch(1); + when(agentService.chatWithUsage(anyLong(), anyString(), anyString(), any())) + .thenAnswer(invocation -> { + childStarted.countDown(); + new CountDownLatch(1).await(5, TimeUnit.SECONDS); + return AgentService.ChatResult.contentOnly("late"); + }); + + tool.delegateAsync("Researcher", "slow task", null, 1, + makeCtx("user-1", "parent-conv-1")); + Callable work = workCaptor.getValue(); + String childConversationId = objectMapper.readTree(requestCaptor.getValue()) + .path("childConversationId").asText(); + + assertThatThrownBy(work::call) + .isInstanceOf(java.util.concurrent.TimeoutException.class) + .hasMessageContaining("timed out after 1 seconds"); + assertThat(childStarted.getCount()).isZero(); + verify(streamTracker).register(childConversationId); + verify(streamTracker).requestStop(childConversationId); + verify(streamTracker).complete(childConversationId); + verify(subagentRegistry).unregister("sa-timeout"); + } + // ---------- taskOutput ---------- @Test