From 81c6f4aece381743e68f6fc28670bae292c244ba Mon Sep 17 00:00:00 2001 From: mateaix <7333791@qq.com> Date: Sun, 16 Aug 2026 16:26:10 +0800 Subject: [PATCH] fix(chat): gracefully terminate active tool runs Stopping a conversation previously disposed the outer reactive stream without reliably cancelling synchronous tool callbacks running on worker threads. Shell commands, skill scripts, and Playwright sessions could therefore outlive the visible chat turn. Propagate cancellation through run-scoped hooks, interrupt active tool execution and parallel batches, terminate subprocess trees, close per-conversation browser sessions, and wait briefly for final stream persistence before acknowledging Stop. Expose an explicit interrupting state in the chat input to block duplicate stop clicks and show progress, with regression coverage for cancelling an in-flight synchronous tool callback. --- .../graph/executor/ToolExecutionExecutor.java | 52 +++++++ .../vip/mate/channel/web/ChatController.java | 5 + .../mate/channel/web/ChatStreamTracker.java | 140 +++++++++++++++++- .../runtime/SkillScriptExecutionService.java | 13 +- .../vip/mate/tool/builtin/BrowserUseTool.java | 8 + .../mate/tool/builtin/ShellExecuteTool.java | 17 ++- ...ToolExecutionExecutorCancellationTest.java | 84 +++++++++++ mateclaw-ui/src/components/chat/ChatInput.vue | 26 +++- mateclaw-ui/src/composables/chat/useChat.ts | 23 ++- 9 files changed, 352 insertions(+), 16 deletions(-) create mode 100644 mateclaw-server/src/test/java/vip/mate/agent/graph/executor/ToolExecutionExecutorCancellationTest.java diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java index 12c767ac..4500618c 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java @@ -734,7 +734,12 @@ public class ToolExecutionExecutor { return new ToolResponseMessage.ToolResponse(toolCall.id(), toolName, msg); } + Thread executionThread = Thread.currentThread(); + Runnable removeCancellationHook = streamTracker != null + ? streamTracker.registerCancellationHook(conversationId, executionThread::interrupt) + : () -> { }; try { + throwIfStopRequested(conversationId); log.info("[ToolExecutor] Executing pre-approved tool: {}", toolName); // RFC-063r §2.5: forward ToolContext so the pre-approved tool can // still observe the originating ChatOrigin (channel/workspace). @@ -745,6 +750,7 @@ public class ToolExecutionExecutor { .withConversationId(conversationId) .withWorkspace(null, workspaceBasePath); String result = callback.call(callArguments, toolContextWithScopedCatalog(replayOrigin)); + throwIfStopRequested(conversationId); int rawLen = result != null ? result.length() : 0; // RFC-052: pre-approved tool may itself be returnDirect — in that @@ -779,6 +785,8 @@ public class ToolExecutionExecutor { // leaving the broadcast tool-result panel unchanged. return new ToolResponseMessage.ToolResponse( toolCall.id(), toolName, withProductCardDirective(toolName, result != null ? result : "")); + } catch (CancellationException e) { + throw e; } catch (Exception e) { log.error("[ToolExecutor] Pre-approved tool {} failed: {}", toolName, e.getMessage()); String safeError = isReturnDirect(callback) @@ -786,6 +794,11 @@ public class ToolExecutionExecutor { : "Tool execution failed: " + e.getMessage(); events.add(GraphEventPublisher.toolComplete(toolCall.id(), toolName, safeError, false)); return new ToolResponseMessage.ToolResponse(toolCall.id(), toolName, safeError); + } finally { + removeCancellationHook.run(); + if (streamTracker != null && streamTracker.isStopRequested(conversationId)) { + Thread.interrupted(); + } } } @@ -826,6 +839,7 @@ public class ToolExecutionExecutor { List> batches = buildExecutionBatches(preparedCalls); for (List batch : batches) { + throwIfStopRequested(preparedCalls.isEmpty() ? null : preparedCalls.get(0).conversationId); if (batch.size() == 1) { // 单个工具（safe 或 unsafe），直接执行 PreparedToolCall pc = batch.get(0); @@ -893,13 +907,26 @@ public class ToolExecutionExecutor { // 等待所有并行工具完成，按原始顺序填入结果 for (var entry : futures.entrySet()) { try { + String conversationId = batch.isEmpty() ? null : batch.get(0).conversationId; + if (streamTracker != null && streamTracker.isStopRequested(conversationId)) { + futures.values().forEach(future -> future.cancel(true)); + throw new CancellationException("Stream stopped by user during tool execution"); + } // 按工具名查找配置的超时时间 PreparedToolCall matchedPc = batch.stream() .filter(p -> p.resultIndex == entry.getKey()).findFirst().orElse(null); long timeoutMs = getToolTimeoutMs(matchedPc != null ? matchedPc.toolCall.name() : null); ToolResponseMessage.ToolResponse response = entry.getValue().get(timeoutMs, TimeUnit.MILLISECONDS); allResponses.set(entry.getKey(), response); + } catch (CancellationException e) { + futures.values().forEach(future -> future.cancel(true)); + throw e; } catch (Exception e) { + String conversationId = batch.isEmpty() ? null : batch.get(0).conversationId; + if (streamTracker != null && streamTracker.isStopRequested(conversationId)) { + futures.values().forEach(future -> future.cancel(true)); + throw new CancellationException("Stream stopped by user during tool execution"); + } // 超时或异常 — 填入错误响应 PreparedToolCall pc = batch.stream() .filter(p -> p.resultIndex == entry.getKey()) @@ -920,7 +947,12 @@ public class ToolExecutionExecutor { List events, List directOutputs) { String toolName = pc.toolCall.name(); + Thread executionThread = Thread.currentThread(); + Runnable removeCancellationHook = streamTracker != null + ? streamTracker.registerCancellationHook(pc.conversationId, executionThread::interrupt) + : () -> { }; try { + throwIfStopRequested(pc.conversationId); if (streamTracker != null) { streamTracker.updateRunningTool(pc.conversationId, toolName); streamTracker.broadcastObject(pc.conversationId, GraphEventPublisher.EVENT_TOOL_START, @@ -956,6 +988,7 @@ public class ToolExecutionExecutor { } result = pc.callback.call(pc.arguments, toolContext); + throwIfStopRequested(pc.conversationId); } finally { if (progressToken != null) { progressContext.remove(progressToken); @@ -1035,6 +1068,11 @@ public class ToolExecutionExecutor { return new ToolResponseMessage.ToolResponse( pc.toolCall.id(), pc.responseName, withProductCardDirective(toolName, result != null ? result : "")); + } catch (CancellationException e) { + if (streamTracker != null) { + streamTracker.updateRunningTool(pc.conversationId, null); + } + throw e; } catch (Exception e) { log.error("[ToolExecutor] Tool {} execution failed: {}", toolName, e.getMessage(), e); // RFC-052: for returnDirect tools, even the error message is @@ -1053,6 +1091,20 @@ public class ToolExecutionExecutor { } return new ToolResponseMessage.ToolResponse( pc.toolCall.id(), pc.responseName, reportedError); + } finally { + removeCancellationHook.run(); + // Virtual-thread workers are not reused, but single/unsafe calls + // can execute on a Reactor worker. Do not leak Stop's interrupt bit + // into unrelated work scheduled on that carrier. + if (streamTracker != null && streamTracker.isStopRequested(pc.conversationId)) { + Thread.interrupted(); + } + } + } + + private void throwIfStopRequested(String conversationId) { + if (streamTracker != null && streamTracker.isStopRequested(conversationId)) { + throw new CancellationException("Stream stopped by user during tool execution"); } } diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java index e5559544..10af9dbe 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java @@ -1027,6 +1027,10 @@ public class ChatController { return R.fail(403, "无权操作该会话"); } boolean stopped = streamTracker.requestStop(conversationId); + // Acknowledge only after the cancellation path had a chance to drain + // and persist its partial assistant message. Bound the wait so a + // genuinely non-cooperative third-party tool cannot pin the HTTP call. + boolean terminationConfirmed = !stopped || streamTracker.awaitTermination(conversationId, 2000L); // Sweep ghost approvals — workflow.denyAllByConversation owns DB + metadata + memory // atomically; we only need to broadcast SSE events on the resulting outcomes. @@ -1045,6 +1049,7 @@ public class ChatController { conversationId, username, stopped, denied.size(), messagesRewritten); return R.ok(Map.of( "stopped", stopped, + "terminationConfirmed", terminationConfirmed, "ghostPendingsCleared", denied.size(), "messagesRewritten", messagesRewritten )); diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java index 43a9bde7..7925c12d 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java @@ -160,6 +160,16 @@ public class ChatStreamTracker { volatile Disposable disposable; /** 停止标志：requestStop() 设为 true，各图节点和 LLM 调用检查此标志以提前退出 */ final AtomicBoolean stopRequested = new AtomicBoolean(false); + /** + * Cancellation hooks owned by work that has escaped the Reactor + * subscription (most notably synchronous ToolCallback invocations). + * Guarded by {@link #lock}; requestStop snapshots and invokes them + * outside the lock so a hook may safely deregister itself. + */ + final java.util.Set cancellationHooks = new java.util.HashSet<>(); + /** Completed only after the run's finalization path has drained. */ + final java.util.concurrent.CompletableFuture termination = + new java.util.concurrent.CompletableFuture<>(); /** * 当前活跃的 Flux 数量（原始流 + 审批 Replay 流共享同一个 RunState）。 * complete() 仅在计数归零时才真正移除 RunState，防止 Replay 仍在运行时被原始流的完成误删。 @@ -548,6 +558,46 @@ public class ChatStreamTracker { } } + /** + * Register cancellation for work performed outside the run's Reactor + * subscription. The returned handle is idempotent and must be closed when + * that work finishes. If Stop already won the race, the hook is invoked + * immediately instead of being registered. + */ + public Runnable registerCancellationHook(String conversationId, Runnable hook) { + if (conversationId == null || hook == null) { + return () -> { }; + } + RunState state = runs.get(conversationId); + if (state == null) { + return () -> { }; + } + boolean cancelImmediately; + synchronized (state.lock) { + cancelImmediately = !isCurrent(state) || state.done || state.stopRequested.get(); + if (!cancelImmediately) { + state.cancellationHooks.add(hook); + } + } + if (cancelImmediately) { + invokeCancellationHook(conversationId, hook); + return () -> { }; + } + return () -> { + synchronized (state.lock) { + state.cancellationHooks.remove(hook); + } + }; + } + + private void invokeCancellationHook(String conversationId, Runnable hook) { + try { + hook.run(); + } catch (Exception e) { + log.warn("Cancellation hook failed for {}: {}", conversationId, e.getMessage()); + } + } + /** * Register an emergency-save callback for this run, invoked from {@link #onShutdown()} * before the JVM tears down. The callback should snapshot the current accumulator @@ -581,12 +631,35 @@ public class ChatStreamTracker { */ public boolean requestStop(String conversationId) { RunState state = runs.get(conversationId); - if (state == null || state.done) { - return false; + if (state == null) return false; + + final boolean firstRequest; + final Disposable d; + final List hooks; + synchronized (state.lock) { + if (!isCurrent(state) || state.done) return false; + // Set the flag before taking the hook snapshot. A tool entering + // concurrently will then self-cancel in registerCancellationHook. + firstRequest = !state.stopRequested.getAndSet(true); + state.currentPhase = "interrupting"; + state.runningToolName = null; + d = state.disposable; + hooks = new ArrayList<>(state.cancellationHooks); + state.cancellationHooks.clear(); + } + + // Let the UI render an explicit transition before cancellation closes + // the stream. This mirrors qwenpaw's cancel envelope instead of making + // the Stop button look unresponsive until final persistence finishes. + broadcastObject(conversationId, "phase", Map.of( + "phase", "interrupting", + "timestamp", System.currentTimeMillis())); + + // Disposing the Flux alone cannot stop a synchronous callback already + // running on another thread. Cancel those escaped executions first. + for (Runnable hook : hooks) { + invokeCancellationHook(conversationId, hook); } - // 设置停止标志，图节点和 LLM 调用会检查此标志以提前退出 - boolean firstRequest = !state.stopRequested.getAndSet(true); - Disposable d = state.disposable; if (d != null && !d.isDisposed()) { d.dispose(); log.info("Stream stopped via requestStop: {}", conversationId); @@ -604,6 +677,23 @@ public class ChatStreamTracker { return state != null && state.stopRequested.get(); } + /** + * Wait briefly for cancellation finalization (partial-message persistence, + * done envelope, and lifecycle cleanup). This gives Stop callers the same + * acknowledgement semantics as qwenpaw's request_stop(), which awaits the + * cancelled task instead of merely sending a signal. + */ + public boolean awaitTermination(String conversationId, long timeoutMillis) { + RunState state = runs.get(conversationId); + if (state == null || state.done) return true; + try { + state.termination.get(Math.max(1L, timeoutMillis), TimeUnit.MILLISECONDS); + return true; + } catch (Exception e) { + return state.done; + } + } + /** * Whether this conversation was force-recycled by an admin within the * recycle marker's TTL ({@link #DONE_RETENTION_MS}). The SSE doOn* @@ -1195,6 +1285,8 @@ public class ChatStreamTracker { return false; } state.done = true; + state.cancellationHooks.clear(); + state.termination.complete(null); oldHeartbeat = state.heartbeatFuture; state.heartbeatFuture = null; } @@ -1238,6 +1330,8 @@ public class ChatStreamTracker { // 最后一个 Flux：在同一个锁内消费排队消息（取队首） consumed = state.messageQueue.poll(); state.done = true; + state.cancellationHooks.clear(); + state.termination.complete(null); oldHeartbeat = state.heartbeatFuture; state.heartbeatFuture = null; } @@ -1913,6 +2007,15 @@ public class ChatStreamTracker { entry.getKey(), ex.getMessage()); } try { + state.stopRequested.set(true); + List hooks; + synchronized (state.lock) { + hooks = new ArrayList<>(state.cancellationHooks); + state.cancellationHooks.clear(); + } + for (Runnable hook : hooks) { + invokeCancellationHook(entry.getKey(), hook); + } Disposable d = state.disposable; if (d != null && !d.isDisposed()) { d.dispose(); @@ -1922,6 +2025,7 @@ public class ChatStreamTracker { entry.getKey(), ex.getMessage()); } } finally { + state.termination.complete(null); mappingRemoved = runs.remove(entry.getKey(), state); reclaimed++; if (mappingRemoved) { @@ -1994,6 +2098,15 @@ public class ChatStreamTracker { log.error("[ChatStreamTracker] Emergency save failed for {}: {}", cid, e.getMessage(), e); } + state.stopRequested.set(true); + List hooks; + synchronized (state.lock) { + hooks = new ArrayList<>(state.cancellationHooks); + state.cancellationHooks.clear(); + } + for (Runnable hook : hooks) { + invokeCancellationHook(cid, hook); + } try { Disposable d = state.disposable; if (d != null && !d.isDisposed()) { @@ -2003,6 +2116,7 @@ public class ChatStreamTracker { log.warn("[ChatStreamTracker] Disposable.dispose failed for {}: {}", cid, e.getMessage()); } + state.termination.complete(null); } } @@ -2160,9 +2274,18 @@ public class ChatStreamTracker { } } try { - state.stopRequested.set(true); - state.interruptType = InterruptType.USER_STOP; - Disposable d = state.disposable; + final Disposable d; + final List hooks; + synchronized (state.lock) { + state.stopRequested.set(true); + state.interruptType = InterruptType.USER_STOP; + d = state.disposable; + hooks = new ArrayList<>(state.cancellationHooks); + state.cancellationHooks.clear(); + } + for (Runnable hook : hooks) { + invokeCancellationHook(conversationId, hook); + } if (d != null && !d.isDisposed()) { d.dispose(); } @@ -2171,6 +2294,7 @@ public class ChatStreamTracker { } try { state.done = true; + state.termination.complete(null); stopHeartbeat(conversationId); } catch (Exception e) { log.warn("forceRecycle: heartbeat stop failed for {}: {}", conversationId, e.getMessage()); diff --git a/mateclaw-server/src/main/java/vip/mate/skill/runtime/SkillScriptExecutionService.java b/mateclaw-server/src/main/java/vip/mate/skill/runtime/SkillScriptExecutionService.java index 39c025b5..78732e4f 100644 --- a/mateclaw-server/src/main/java/vip/mate/skill/runtime/SkillScriptExecutionService.java +++ b/mateclaw-server/src/main/java/vip/mate/skill/runtime/SkillScriptExecutionService.java @@ -168,6 +168,7 @@ public class SkillScriptExecutionService { Path stdoutFile = null; Path stderrFile = null; + Process process = null; try { // 构建命令（结构化参数，避免 shell 注入） @@ -242,7 +243,7 @@ public class SkillScriptExecutionService { } injectPipMirrorEnv(pb); - Process process = pb.start(); + process = pb.start(); boolean finished = process.waitFor(timeoutSeconds, TimeUnit.SECONDS); if (!finished) { @@ -259,6 +260,16 @@ public class SkillScriptExecutionService { String stderr = readFileTruncated(stderrFile, MAX_OUTPUT_BYTES); return new ScriptResult(exitCode, stdout, stderr); + } catch (InterruptedException e) { + // Stop must terminate the OS process as well as the Java wait. + // Without this, cancelling the outer chat stream leaves skill + // scripts running until their normal timeout. + if (process != null && process.isAlive()) { + killProcess(process); + } + Thread.currentThread().interrupt(); + log.info("Skill script interrupted by conversation cancellation: {}", scriptPath); + return ScriptResult.error(-1, "Execution cancelled by user"); } catch (Exception e) { log.error("Failed to execute script {}: {}", scriptPath, e.getMessage()); return ScriptResult.error(-1, "Execution error: " + e.getMessage()); diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/BrowserUseTool.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/BrowserUseTool.java index 13095b5a..be754ddf 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/BrowserUseTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/BrowserUseTool.java @@ -248,6 +248,12 @@ public class BrowserUseTool { String conversationId = ToolExecutionContext.conversationId(ctx); String sessionKey = (conversationId != null && !conversationId.isBlank()) ? conversationId : "default"; + // Closing the per-conversation session aborts Playwright's native + // wait/navigation even when a Java thread interrupt alone is not + // observed by the driver transport. + Runnable removeCancellationHook = streamTracker != null && conversationId != null + ? streamTracker.registerCancellationHook(conversationId, () -> doStop(sessionKey)) + : () -> { }; log.info("[BrowserUse] action={}, session={}, url={}, selector={}, headed={}, cdpPort={}", action, sessionKey, url, selector, headed, cdpPort); @@ -283,6 +289,8 @@ public class BrowserUseTool { } finally { currentToolContext.remove(); } + } finally { + removeCancellationHook.run(); } } diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/ShellExecuteTool.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/ShellExecuteTool.java index c8224a2a..943d16a1 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/ShellExecuteTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/ShellExecuteTool.java @@ -90,6 +90,7 @@ public class ShellExecuteTool { Path stdoutFile = null; Path stderrFile = null; + Process process = null; try { // 处理命令中的嵌入换行符（LLM 生成的 JSON 解码后可能包含真实换行） @@ -113,7 +114,7 @@ public class ShellExecuteTool { pb.redirectError(stderrFile.toFile()); long runStart = System.currentTimeMillis(); - Process process = pb.start(); + process = pb.start(); boolean completed = process.waitFor(timeout, TimeUnit.SECONDS); @@ -146,6 +147,20 @@ public class ShellExecuteTool { } } + } catch (InterruptedException e) { + // A conversation Stop interrupts the active tool thread. Kill the + // subprocess tree before returning control; otherwise the Flux is + // gone but the shell command keeps running in the background. + if (process != null && process.isAlive()) { + killProcessTree(process); + } + Thread.currentThread().interrupt(); + log.info("[ShellExecute] Command interrupted by cancellation"); + result.set("exitCode", -1); + result.set("stdout", ""); + result.set("stderr", "Command cancelled by user"); + result.set("timedOut", false); + result.set("cancelled", true); } catch (Exception e) { log.error("[ShellExecute] Command execution failed: {}", e.getMessage(), e); result.set("exitCode", -1); diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/executor/ToolExecutionExecutorCancellationTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/executor/ToolExecutionExecutorCancellationTest.java new file mode 100644 index 00000000..74ced4a1 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/executor/ToolExecutionExecutorCancellationTest.java @@ -0,0 +1,84 @@ +package vip.mate.agent.graph.executor; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.ai.chat.model.ToolContext; +import org.springframework.ai.tool.ToolCallback; +import org.springframework.ai.tool.definition.ToolDefinition; +import org.springframework.ai.tool.metadata.ToolMetadata; +import vip.mate.agent.AgentToolSet; +import vip.mate.channel.web.ChatStreamTracker; +import vip.mate.tool.guard.ToolGuard; +import vip.mate.tool.guard.ToolGuardResult; + +import java.util.List; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class ToolExecutionExecutorCancellationTest { + + @Test + @DisplayName("Stop interrupts an in-flight synchronous tool instead of only disposing the outer stream") + void stopInterruptsActiveToolThread() throws Exception { + String conversationId = "cancel-tool-conversation"; + ChatStreamTracker tracker = new ChatStreamTracker(new ObjectMapper()); + tracker.register(conversationId); + + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch interrupted = new CountDownLatch(1); + ToolCallback blockingTool = blockingTool(entered, interrupted); + AgentToolSet toolSet = AgentToolSet.fromCallbacks(List.of(), List.of(blockingTool)); + ToolGuard alwaysAllow = (name, arguments) -> ToolGuardResult.allow(); + ToolExecutionExecutor executor = new ToolExecutionExecutor(toolSet, alwaysAllow, null, tracker); + AssistantMessage.ToolCall call = new AssistantMessage.ToolCall( + "call-1", "function", "blocking_tool", "{}"); + + CompletableFuture execution = CompletableFuture.runAsync(() -> + executor.execute(List.of(call), conversationId, "agent-1", false)); + + assertTrue(entered.await(2, TimeUnit.SECONDS), "tool callback should have started"); + assertTrue(tracker.requestStop(conversationId), "active run should accept Stop"); + assertTrue(interrupted.await(2, TimeUnit.SECONDS), "Stop must interrupt the tool thread"); + + ExecutionException error = assertThrows(ExecutionException.class, + () -> execution.get(2, TimeUnit.SECONDS)); + assertInstanceOf(CancellationException.class, error.getCause()); + } + + private static ToolCallback blockingTool(CountDownLatch entered, CountDownLatch interrupted) { + ToolDefinition definition = ToolDefinition.builder() + .name("blocking_tool") + .description("blocks until interrupted") + .inputSchema("{\"type\":\"object\",\"properties\":{}}") + .build(); + return new ToolCallback() { + @Override public ToolDefinition getToolDefinition() { return definition; } + @Override public ToolMetadata getToolMetadata() { + return ToolMetadata.builder().returnDirect(false).build(); + } + @Override public String call(String arguments) { return runBlocking(); } + @Override public String call(String arguments, ToolContext toolContext) { return runBlocking(); } + + private String runBlocking() { + entered.countDown(); + try { + Thread.sleep(TimeUnit.MINUTES.toMillis(1)); + return "unexpected completion"; + } catch (InterruptedException e) { + interrupted.countDown(); + Thread.currentThread().interrupt(); + return "cancelled"; + } + } + }; + } +} diff --git a/mateclaw-ui/src/components/chat/ChatInput.vue b/mateclaw-ui/src/components/chat/ChatInput.vue index da7c35fa..47253025 100644 --- a/mateclaw-ui/src/components/chat/ChatInput.vue +++ b/mateclaw-ui/src/components/chat/ChatInput.vue @@ -204,11 +204,13 @@ type="button" class="action-btn send-btn" :class="sendBtnClass" - :disabled="!canSend && !loading" + :disabled="props.streamPhase === 'interrupting' || (!canSend && !loading)" + :title="props.streamPhase === 'interrupting' ? t('chat.streamInterrupting') : undefined" @click="handleSubmit" > - + + @@ -241,7 +243,7 @@