This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 5ba3f0b9 [runtime] Do not retry or persist interrupted tool calls on
cancellation (#1111)
5ba3f0b9 is described below
commit 5ba3f0b92584c969be6be173eb3c0a0427700beb
Author: Ashfaq <[email protected]>
AuthorDate: Wed Sep 16 12:49:09 2026 +0530
[runtime] Do not retry or persist interrupted tool calls on cancellation
(#1111)
Generated-by: Claude Code 2.1.272 (Claude Sonnet 5)
---
.../flink/agents/plan/actions/ToolCallAction.java | 25 ++++++-
.../agents/plan/actions/ToolCallActionTest.java | 77 +++++++++++++++++++-
.../runtime/context/JavaRunnerContextImpl.java | 24 ++++++-
.../agents/runtime/context/RunnerContextImpl.java | 6 ++
...vaRunnerContextImplDurableExecuteAsyncTest.java | 81 ++++++++++++++++++++++
.../RunnerContextImplDurableExecuteTest.java | 40 +++++++++++
6 files changed, 247 insertions(+), 6 deletions(-)
diff --git
a/plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java
b/plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java
index a67571a8..e9ea9573 100644
---
a/plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java
+++
b/plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java
@@ -64,7 +64,8 @@ public class ToolCallAction {
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 class ToolCallAction {
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 @@ public class ToolCallAction {
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 @@ public class ToolCallAction {
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 @@ public class ToolCallAction {
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 @@ public class ToolCallAction {
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(
diff --git
a/plan/src/test/java/org/apache/flink/agents/plan/actions/ToolCallActionTest.java
b/plan/src/test/java/org/apache/flink/agents/plan/actions/ToolCallActionTest.java
index 57e152a9..8be03fdc 100644
---
a/plan/src/test/java/org/apache/flink/agents/plan/actions/ToolCallActionTest.java
+++
b/plan/src/test/java/org/apache/flink/agents/plan/actions/ToolCallActionTest.java
@@ -51,6 +51,7 @@ import java.util.Map;
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 class ToolCallActionTest {
}
@Test
- void processToolRequestUsesFallbackWhenMissingToolErrorHasNoMessage() {
+ void processToolRequestUsesFallbackWhenMissingToolErrorHasNoMessage()
throws Exception {
FakeRunnerContext ctx =
new FakeRunnerContext() {
@Override
@@ -560,6 +561,80 @@ public class ToolCallActionTest {
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")));
}
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java
index 5b028012..7e241596 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java
@@ -192,6 +192,15 @@ public class JavaRunnerContextImpl extends
RunnerContextImpl {
plan.outcomes.set(callIndex, outcome);
continue;
}
+ if (outcome.isFailure() && outcome.getError() instanceof
InterruptedException) {
+ // A cancellation signal a batch executor caught and folded
into a failed
+ // Outcome, not a genuine tool failure: leave this slot and
every slot after it
+ // in the batch unfinalized (pending), so recovery re-executes
them, and
+ // propagate immediately instead of persisting it as a durable
failure. Slots
+ // finalized earlier in this loop keep their persisted outcome.
+ Thread.currentThread().interrupt();
+ throw (InterruptedException) outcome.getError();
+ }
try {
finalizeCallAt(
base + callIndex,
@@ -218,12 +227,23 @@ public class JavaRunnerContextImpl extends
RunnerContextImpl {
}
}
- private <T> List<Outcome<T>>
executeAllWithoutDurableState(List<DurableCallable<T>> callables) {
+ private <T> List<Outcome<T>>
executeAllWithoutDurableState(List<DurableCallable<T>> callables)
+ throws InterruptedException {
List<Callable<T>> suppliers = new ArrayList<>();
for (DurableCallable<T> callable : callables) {
suppliers.add(callable::call);
}
- return executeOutcomeSuppliers(suppliers).getOutcomes();
+ List<Outcome<T>> outcomes =
executeOutcomeSuppliers(suppliers).getOutcomes();
+ for (Outcome<T> outcome : outcomes) {
+ if (outcome.isFailure() && outcome.getError() instanceof
InterruptedException) {
+ // Without a durable store there is nothing to leave pending,
but a cancellation
+ // signal folded into a failed Outcome must still propagate as
an interruption
+ // instead of being handed to the caller as an ordinary tool
failure.
+ Thread.currentThread().interrupt();
+ throw (InterruptedException) outcome.getError();
+ }
+ }
+ return outcomes;
}
private <T> BatchExecutionResult<T>
executeOutcomeSuppliers(List<Callable<T>> suppliers) {
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
index fe1c4ccf..e626afaa 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
@@ -502,6 +502,12 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
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));
}
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java
index adc75fa9..2aa26a73 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java
@@ -527,6 +527,87 @@ class JavaRunnerContextImplDurableExecuteAsyncTest {
executor.close();
}
+ @Test
+ void
testDurableExecuteAllAsyncInterruptionLeavesRemainingSlotsPendingAndPropagates()
+ throws Exception {
+ InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
+ JavaRunnerContextImpl context = createContext(new ActionState(null),
executor);
+ TestDurableCallable<String> first =
+ new TestDurableCallable<>("batch-1", String.class, () ->
"one");
+ TestDurableCallable<String> second =
+ new TestDurableCallable<>(
+ "batch-2",
+ String.class,
+ () -> {
+ throw new InterruptedException("cancelled");
+ });
+ TestDurableCallable<String> third =
+ new TestDurableCallable<>("batch-3", String.class, () ->
"three");
+ Thread.interrupted(); // clear any stray interrupt flag left over from
another test
+
+ InterruptedException thrown =
+ assertThrows(
+ InterruptedException.class,
+ () -> context.durableExecuteAllAsync(List.of(first,
second, third)));
+
+ assertEquals("cancelled", thrown.getMessage());
+ assertTrue(
+ Thread.interrupted(),
+ "the calling thread's interrupt flag must be restored, not
swallowed");
+ List<CallResult> persisted =
+
context.getDurableExecutionContext().getActionState().getCallResults();
+ assertEquals("batch-1", persisted.get(0).getFunctionId());
+ assertTrue(
+ persisted.get(0).isSuccess(),
+ "the slot finalized before the interruption was observed must
keep its outcome");
+ assertTrue(
+ persisted.get(1).isPending(),
+ "the interrupted slot must not be persisted as a durable tool
failure");
+ assertTrue(
+ persisted.get(2).isPending(),
+ "slots after the interrupted one must stay pending, not be
finalized");
+ assertEquals(
+ 0,
+ context.getDurableExecutionContext().getCurrentCallIndex(),
+ "the call index must not advance past an interrupted batch, so
recovery re-reads"
+ + " the pending slots instead of skipping them");
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncWithoutDurableStateInterruptionPropagates()
throws Exception {
+ InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
+ JavaRunnerContextImpl context =
+ new JavaRunnerContextImpl(
+ metricGroup,
+ () -> {},
+ new AgentPlan(new HashMap<>(), new HashMap<>()),
+ null,
+ "test-job",
+ executor);
+ context.setContinuationContext(new ContinuationContext());
+ TestDurableCallable<String> first =
+ new TestDurableCallable<>("batch-1", String.class, () ->
"one");
+ TestDurableCallable<String> second =
+ new TestDurableCallable<>(
+ "batch-2",
+ String.class,
+ () -> {
+ throw new InterruptedException("cancelled");
+ });
+ Thread.interrupted(); // clear any stray interrupt flag left over from
another test
+
+ InterruptedException thrown =
+ assertThrows(
+ InterruptedException.class,
+ () -> context.durableExecuteAllAsync(List.of(first,
second)));
+
+ assertEquals("cancelled", thrown.getMessage());
+ assertTrue(
+ Thread.interrupted(),
+ "the calling thread's interrupt flag must be restored, not
swallowed, even"
+ + " without a durable store to persist pending slots
into");
+ }
+
@Test
void
testDurableExecuteAllAsyncFinalizeFailureReturnsOutcomeAndKeepsSlotPending()
throws Exception {
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/RunnerContextImplDurableExecuteTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/RunnerContextImplDurableExecuteTest.java
index 816b874d..228eee78 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/RunnerContextImplDurableExecuteTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/RunnerContextImplDurableExecuteTest.java
@@ -31,6 +31,7 @@ import org.junit.jupiter.api.BeforeEach;
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 @@ class RunnerContextImplDurableExecuteTest {
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));
