[runtime] Do not retry or persist interrupted tool calls on cancellation

plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java

@@ -64,7 +64,8 @@ public static Action getToolCallAction() throws Exception {
                 List.of(ToolRequestEvent.EVENT_TYPE));
     }
 
-    public static void processToolRequest(Event event, RunnerContext ctx) {
+    public static void processToolRequest(Event event, RunnerContext ctx)
+            throws InterruptedException {
         ToolRequestEvent toolRequest = ToolRequestEvent.fromEvent(event);
         boolean toolCallAsync = ctx.getConfig().get(AgentExecutionOptions.TOOL_CALL_ASYNC);
         int toolCallParallelism = ctx.getConfig().get(AgentExecutionOptions.TOOL_CALL_PARALLELISM);
@@ -76,6 +77,10 @@ public static void processToolRequest(Event event, RunnerContext ctx) {
         List<ToolCallExecution> executions =
                 buildToolCallExecutions(toolRequest, ctx, externalIds, success, error, responses);
 
+        // executeParallel/executeSequentially let InterruptedException propagate rather than
+        // recording it as a tool error, so a cancellation here skips sendEvent below entirely:
+        // no ToolResponseEvent goes out, no further chat call gets driven off a cancelled tool
+        // call, and the action is never persisted as completed on the back of it.
         if (toolCallAsync && toolCallParallelism > 1 && executions.size() > 1) {
             executeParallel(executions, ctx, success, error, responses);
         } else {
@@ -196,7 +201,8 @@ private static void executeParallel(
             RunnerContext ctx,
             Map<String, Boolean> success,
             Map<String, String> error,
-            Map<String, ToolResponse> responses) {
+            Map<String, ToolResponse> responses)
+            throws InterruptedException {
         List<DurableCallable<ToolResponse>> callables = new ArrayList<>(executions.size());
         for (ToolCallExecution execution : executions) {
             callables.add(execution.callable);
@@ -206,6 +212,12 @@ private static void executeParallel(
             for (int i = 0; i < outcomes.size(); i++) {
                 recordOutcome(executions.get(i), outcomes.get(i), ctx, success, error, responses);
             }
+        } catch (InterruptedException e) {
+            // A cancellation signal, not a batch failure: propagate immediately instead of
+            // recording every execution as a tool error and letting the caller send a
+            // ToolResponseEvent that drives the action loop onward.
+            Thread.currentThread().interrupt();
+            throw e;
         } catch (Exception e) {
             for (ToolCallExecution execution : executions) {
                 recordExecutionException(execution, e, success, error, responses);
@@ -230,7 +242,8 @@ private static void executeSequentially(
             RunnerContext ctx,
             Map<String, Boolean> success,
             Map<String, String> error,
-            Map<String, ToolResponse> responses) {
+            Map<String, ToolResponse> responses)
+            throws InterruptedException {
         for (ToolCallExecution execution : executions) {
             try {
                 ToolResponse response =
@@ -253,6 +266,12 @@ private static void executeSequentially(
                             execution.name,
                             execution.entityMetadata);
                 }
+            } catch (InterruptedException e) {
+                // A cancellation signal, not a tool failure: propagate immediately instead of
+                // recording it as a tool error and letting the loop move on to (or past) the
+                // remaining executions and the caller send a ToolResponseEvent for it.
+                Thread.currentThread().interrupt();
+                throw e;
             } catch (Exception e) {
                 recordExecutionException(execution, e, success, error, responses);
                 ExecutionReporters.failed(

plan/src/test/java/org/apache/flink/agents/plan/actions/ToolCallActionTest.java

@@ -51,6 +51,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
 public class ToolCallActionTest {
 
@@ -428,7 +429,7 @@ public Resource getResource(String name, ResourceType type) {
     }
 
     @Test
-    void processToolRequestUsesFallbackWhenMissingToolErrorHasNoMessage() {
+    void processToolRequestUsesFallbackWhenMissingToolErrorHasNoMessage() throws Exception {
         FakeRunnerContext ctx =
                 new FakeRunnerContext() {
                     @Override
@@ -560,6 +561,80 @@ public <T> List<Outcome<T>> durableExecuteAllAsync(
         assertThat(response.getError()).containsEntry("call-2", "persist failed");
     }
 
+    @Test
+    void processToolRequestPropagatesInterruptionInsteadOfRecordingToolError() {
+        FakeRunnerContext ctx =
+                new FakeRunnerContext() {
+                    @Override
+                    public <T> T durableExecute(DurableCallable<T> callable) throws Exception {
+                        throw new InterruptedException("cancelled");
+                    }
+                }.withToolCallAsync(false);
+
+        Thread.interrupted();
+
+        assertThatExceptionOfType(InterruptedException.class)
+                .isThrownBy(
+                        () -> ToolCallAction.processToolRequest(toolRequest("queryOrder"), ctx));
+
+        assertThat(Thread.interrupted()).as("interrupt status should be restored").isTrue();
+        // No ToolResponseEvent should go out for a cancelled call: the loop must not fold the
+        // interruption into a tool-error response and drive a further chat call off it.
+        assertThat(ctx.sentEvents).isEmpty();
+    }
+
+    @Test
+    void processToolRequestStopsSequentialLoopOnFirstInterruptionWithoutRunningLaterTools() {
+        AtomicInteger calls = new AtomicInteger();
+        FakeRunnerContext ctx =
+                new FakeRunnerContext() {
+                    @Override
+                    public <T> T durableExecute(DurableCallable<T> callable) throws Exception {
+                        if (calls.incrementAndGet() == 1) {
+                            throw new InterruptedException("cancelled");
+                        }
+                        return callable.call();
+                    }
+                }.withToolCallAsync(false);
+
+        Thread.interrupted();
+
+        assertThatExceptionOfType(InterruptedException.class)
+                .isThrownBy(
+                        () ->
+                                ToolCallAction.processToolRequest(
+                                        toolRequest("queryOrder", "call-1", "call-2"), ctx));
+
+        Thread.interrupted();
+        assertThat(calls.get())
+                .as("the second tool call must not run once the first is interrupted")
+                .isEqualTo(1);
+        assertThat(ctx.sentEvents).isEmpty();
+    }
+
+    @Test
+    void processToolRequestPropagatesInterruptionFromParallelBatchWithoutRecordingToolErrors() {
+        FakeRunnerContext ctx =
+                new FakeRunnerContext() {
+                    @Override
+                    public <T> List<Outcome<T>> durableExecuteAllAsync(
+                            List<DurableCallable<T>> callables) throws Exception {
+                        throw new InterruptedException("cancelled");
+                    }
+                };
+
+        Thread.interrupted();
+
+        assertThatExceptionOfType(InterruptedException.class)
+                .isThrownBy(
+                        () ->
+                                ToolCallAction.processToolRequest(
+                                        toolRequest("queryOrder", "call-1", "call-2"), ctx));
+
+        assertThat(Thread.interrupted()).as("interrupt status should be restored").isTrue();
+        assertThat(ctx.sentEvents).isEmpty();
+    }
+
     private static ToolRequestEvent toolRequest(String toolName) {
         return new ToolRequestEvent("model", List.of(toolCall(toolName, "call-1", "order-1")));
     }

runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java

@@ -496,6 +496,12 @@ public <T> List<Outcome<T>> durableExecuteAllAsync(List<DurableCallable<T>> call
         for (DurableCallable<T> callable : callables) {
             try {
                 outcomes.add(Outcome.success(durableExecute(callable)));
+            } catch (InterruptedException e) {
+                // A cancellation signal, not a genuine call failure: stop scheduling the
+                // remaining callables and propagate immediately, instead of recording it as a
+                // failed outcome and continuing on to the rest of the batch.
+                Thread.currentThread().interrupt();
+                throw e;
             } catch (Exception e) {
                 outcomes.add(Outcome.failure(e));
             }

runtime/src/test/java/org/apache/flink/agents/runtime/context/RunnerContextImplDurableExecuteTest.java

@@ -31,6 +31,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.HashMap;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.jupiter.api.Assertions.*;
@@ -127,6 +128,45 @@ void testDurableExecuteCompletionOnlyReExecutesPendingSlotDoesNotPersistInterrup
         assertTrue(pending.isPending(), "interrupted pending slot should remain unfinalized");
     }
 
+    @Test
+    void testDurableExecuteAllAsyncStopsBatchOnInterruptionInsteadOfRecordingFailure() {
+        RunnerContextImpl context = createContext(new ActionState(null));
+        TestDurableCallable<String> first =
+                new TestDurableCallable<>("batch-call-1", String.class, () -> "ok");
+        TestDurableCallable<String> second =
+                new TestDurableCallable<>(
+                        "batch-call-2",
+                        String.class,
+                        () -> {
+                            throw new InterruptedException("cancelled");
+                        });
+        TestDurableCallable<String> third =
+                new TestDurableCallable<>(
+                        "batch-call-3",
+                        String.class,
+                        () -> fail("later callables must not run once the batch is interrupted"));
+
+        Thread.interrupted();
+
+        assertThrows(
+                InterruptedException.class,
+                () -> context.durableExecuteAllAsync(List.of(first, second, third)));
+
+        assertTrue(Thread.interrupted(), "interrupt status should be restored on the thread");
+        assertEquals(1, first.getCallCount());
+        assertEquals(1, second.getCallCount());
+        assertEquals(0, third.getCallCount(), "callables after the interrupted one must not run");
+        // batch-call-1 genuinely completed before the interruption, so it's correctly persisted;
+        // batch-call-2's interruption must not be recorded as a failed Outcome, though — it
+        // should propagate out of durableExecuteAllAsync instead, same as any other
+        // durableExecute call, leaving nothing persisted for it.
+        assertEquals(1, context.getDurableExecutionContext().getActionState().getCallResultCount());
+        CallResult persisted =
+                context.getDurableExecutionContext().getActionState().getCallResults().get(0);
+        assertEquals("batch-call-1", persisted.getFunctionId());
+        assertTrue(persisted.isSuccess());
+    }
+
     @Test
     void testDurableExecuteReconcilableSuccessCall() throws Exception {
         RunnerContextImpl context = createContext(new ActionState(null));