weiqingy commented on code in PR #1071:
URL: https://github.com/apache/flink-agents/pull/1071#discussion_r3911067770
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ChatModelInvoker.java:
##########
@@ -182,6 +182,12 @@ public ChatMessage call() throws Exception {
}
return new ChatAttemptResult(
model, chatModel, response, actualRetryCount,
totalWaitTimeSec);
+ } catch (InterruptedException e) {
+ // A cancellation signal, not a model failure: restore the
interrupt status and
+ // propagate immediately so task shutdown isn't delayed by
retry backoff or an
+ // extra model call, regardless of the configured
error-handling strategy.
+ Thread.currentThread().interrupt();
+ throw e;
} catch (Exception e) {
Review Comment:
The new block has no test. Both tests pass `retryWaitIntervalSec = 0` (lines
80 and 112), so the guard on line 202 skips lines 203 to 208.
A cheap deterministic test, in case it helps. If the stub sets the flag
before it throws, `Thread.sleep` throws at once and costs no wall time:
```java
when(ctx.durableExecute(any()))
.thenAnswer(
inv -> {
Thread.currentThread().interrupt();
throw new RuntimeException("transient failure");
});
```
with `RETRY`, `numRetries = 1`, `retryWaitIntervalSec = 1`.
One catch. `assertThrows` and the `times(1)` verify pass on the old code
too, so only `assertTrue(Thread.interrupted())` proves the fix. Is a test here
worth adding?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java:
##########
@@ -576,6 +576,12 @@ protected <T> T durableExecuteCompletionOnly(
Exception exception = null;
try {
result = executionCallable.call();
+ } catch (InterruptedException e) {
+ // A cancellation signal, not a genuine call failure: leave the
durable slot
+ // unfinished so recovery re-executes or reconciles the call
instead of replaying a
+ // stale interruption as a completed success or failure.
+ Thread.currentThread().interrupt();
+ throw e;
Review Comment:
Agreed, leave it out of this PR. I would skip the follow-up issue too. #1042
already has the same fix. It adds `ModelRoutingResolver.isCancellation(...)`
and calls it in `ChatModelAction.processChatRequest` just before the IGNORE
check.
#1042 is also what makes this path reachable. `Strategies.llm(...)` runs a
judge chat call through `chatWithRetries`, so the strategy that does I/O
arrives with its own guard.
It is still open though, so this only holds if it lands as it stands. Does
that look right to you?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java:
##########
@@ -939,6 +945,12 @@ protected <T> T executeAndFinalizeCurrentCall(
Exception exception = null;
try {
result = callSupplier.call();
+ } catch (InterruptedException e) {
+ // A cancellation signal, not a genuine call failure: leave the
pending call
+ // unfinalized so recovery re-executes or reconciles it instead of
replaying a stale
+ // interruption as a completed success or failure.
+ Thread.currentThread().interrupt();
+ throw e;
Review Comment:
Agreed, a separate issue is right. Nothing open covers tool-call
cancellation, so it will not duplicate anything.
One thing that might be worth adding to it. Because the action returns
normally, `ActionExecutionOperator` persists it as completed (line 491), and on
recovery line 437 replays the output events without running the call again.
That is problem 2 in #1070, carried by the action state rather than the durable
slot. The chat path escapes it because the raw `InterruptedException` is
rethrown on line 482, before the persist. Worth folding in?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
