[Feature] Parallel Tool Call Execution

[da-daken]

Linked issue: #925

Purpose of change

Add parallel execution capability for tool calls via a new public API (RunnerContext.durableExecuteAllAsync()), which submits a batch of tool calls concurrently to an async thread pool and returns results in the original order after all complete.
The execution is durable: the state of in-flight calls is persisted, so that upon recovery (e.g., from checkpoint) the system correctly restores and completes the pending calls, ensuring no duplicate or lost invocations.

Tests

• Unit tests to verify the ordering and consistency of parallel execution results.
• Integration tests to validate correctness under concurrent scenarios.
• Fault-tolerance tests to ensure that after a simulated failure and recovery, the state of parallel executions is correctly restored and all results are eventually returned exactly once.

API

• Introduce a new public API RunnerContext.durableExecuteAllAsync() for batch parallel execution of tool calls.
• Expose the parallel execution capability without breaking compatibility with the existing synchronous API.

Documentation

• doc-needed
• doc-not-needed
• doc-included

[weiqingy]

Thanks for pushing this through. A few questions inline.

[weiqingy]

This is the repo's first ConfigOption<Duration>, and I do not think it can be set from any documented route.

AgentConfiguration.get() dispatches isAssignableFrom, String, Integer, Long, Float, Double, Boolean, isEnum(), then throw new ClassCastException. There is no Duration branch, and YAML values arrive as String or Integer. So getConfig().get(TOOL_CALL_BATCH_TIMEOUT) at JavaRunnerContextImpl.java:253 throws for any value a user sets, after :119 has already persisted N PENDING slots, and the catch-all at ToolCallAction.java:171 then reports every tool in the batch failed.

The programmatic route clears get() but not plan serialization: AgentPlan.writeObject uses a bare new ObjectMapper() with no JavaTimeModule, and I reproduced InvalidDefinitionException on java.time.Duration against the pinned jackson-databind. On Python, core_options.py:271 is config_type=int, so the documented 30s raises ValueError mid-action.

CI stays green because JavaRunnerContextImplDurableExecuteAsyncTest.java:324-327 sets the option in-process, where isAssignableFrom short-circuits. And check_java_python_config_options_parity.py:45 adds "java.time.Duration": int to the type table, so the harness now certifies exactly this pair.

ConfigurationUtils.convertValue is already imported in AgentConfiguration for the enum branch and it handles Duration; retyping this option to Long millis would instead let the harness stay strict. Do you have a preference? Asking because the batch deadline is the only bound on a hung tool wedging the fan-in, and it currently ships both default-off and unsettable.

[da-daken]

Thanks for the catch. I missed the YAML case and have now updated the timeout configuration to use the Long type

[weiqingy]

The type change buys more than the YAML fix, for what it's worth — AgentConfiguration.get() has a Long branch at AgentConfiguration.java:142, and Long is Jackson-native, so the plan-serialization and Python routes resolve along with it.

Two smaller things while the option is in flux. short-term-memory.state-ttl.ms (AgentExecutionOptions.java:88) is the existing Long-millis option in this class and puts the unit in the key — since a bare Long drops the unit Duration carried, would tool-call.batch.timeout.ms be worth matching to it while the key is still unreleased?

And a few Duration leftovers will probably want sweeping with the same change: the parity-harness entries added for it (check_java_python_config_options_parity.py:45 and the _java_duration_to_millis branch at :99) are covered by "java.lang.Long": int at :40 once the type moves, and the config table still lists the type as Duration with a -1ms default (configuration.md:137). Any reason to keep those around once the type moves?

[weiqingy]

testToolCallBatchExecutionIsActuallyParallel covers the real java21 path well, fallback branch included, which matters because surefire runs the exploded target/classes and so never loads the META-INF/versions/21 classes. The deadline leg looks like the remaining gap, and this stub is what makes it look covered.

collectTimedOutOutcomes hard-codes "index 0 succeeds, everything else is a TimeoutException" and never reads the timeout argument. So in testDurableExecuteAllAsyncTimeoutKeepsCompletedOutcomes the 10 ms deadline set at :324-327 and the Thread.sleep(100) in the second supplier are both inert, supplier 1 is never invoked, and its getCallCount() is never asserted. A deadline computed with the wrong sign, a cancel on the wrong future, or a barrier that never resolves would all still pass. What it does cover, a timed-out slot persisted FAILED with the cursor still advancing by 2, is worth keeping.

Now that the e2e path exists, would a batch that overruns a short tool-call.batch.timeout be the cheapest way to get getDeadlineNanos and collectBatchOutcomesOnTimeout genuinely exercised? That may have to wait on the Duration conversion on AgentExecutionOptions.java:82, since setting the option is what trips the plan serializer.

[da-daken]

I will add e2e tests to cover realistic timeout scenarios after the previous configuration issue is fixed.

[weiqingy]

Sounds good, and sequencing it after the config fix makes sense. One gap the e2e won't close though: the stub at :404 ignores the timeout argument it's handed, so that assertion passes regardless of the deadline. Worth tightening at the same time?

[weiqingy]

Reserving PENDING for every call in the batch, non-reconcilable ones included, is the intended shape, and it carries the guarantee that such a slot is re-executed on recovery. The batch path honours that at :147-151. The serial path can read those same slots and does not.

Before this PR the two never met. PENDING was written only by durableExecuteWithReconcile (RunnerContextImpl.java:612, :618) and the Python bridge's _prepare_reconciler_execution, both gated on a non-null reconciler(), and that method gates its own cache read on !current.isPending() at :622. The completion-only path never needed that gate, because non-reconcilable calls never wrote a PENDING slot. Now they do: ToolCallAction.java:135 never overrides reconciler() and the interface default is null.

CallResult.matches compares only functionId and argsDigest (CallResult.java:160-162), so a reserved slot reads as a cache hit at RunnerContextImpl.java:757, both payloads are null, and tryGetCachedResult throws NPE on return Optional.of(null) at :575. ToolCallAction.java:192 catches it, so the tool is reported failed rather than re-executed.

Two routes onto the serial path after a mid-batch crash, with the reserved slots still checkpointed. Setting tool-call.parallel=false and restarting is one. The other needs no config change: a trailing tool whose resource fails to resolve on the recovery run is dropped at ToolCallAction.java:114, executions.size() falls to 1, and the > 1 guard at :67 routes to serial.

Treating a PENDING slot as a miss in matchNextOrClearSubsequentCallResult / tryGetCachedResult would extend the same re-execute guarantee to that path. Would that be the right place for it, or is there a reason the serial read should keep seeing a hit?

[da-daken]

Good catch! This is indeed a serious issue. I'll fix the problem where non‑reconcilable ones under the serial path contain pending slots. For recovery reruns, I'll use finalizeCallAt to handle the existing pending slots instead of creating a new one, which would otherwise cause another error.

[weiqingy]

That plan sounds right. One part I wasn't sure it reaches: the read side is status-blind independently of how the slot got written. CallResult.matches (CallResult.java:160) compares only functionId and argsDigest, so a reserved PENDING slot would still read as a hit at the match site even once finalizeCallAt owns the write side. Does the serial path want a status guard there too, or does your fix remove the route that reaches it with a PENDING slot in the first place?

[da-daken]

Yes, I'll mark it as 'no result read' during the read operation and include this in the next commit. I'm a bit tight on time at the moment, so I plan to push it over the weekend. We can revisit it then～

[weiqingy]

Any throw out of durableExecuteAllAsync after partial progress marks every tool failed, including slots already durably finalized SUCCEEDED. The action is then marked completed, so those results never surface on a later run. Lost, not delayed.

The serial path just below degrades per tool; this one is all-or-nothing. Routes in besides the Duration conversion on AgentExecutionOptions.java:82: a JsonProcessingException out of finalizeExecutedOutcomes (JavaRunnerContextImpl.java:211), or a deserialization failure in readTerminalOutcomeAt.

Would recording the batch-level exception only against slots with no outcome yet work here, or is failing the whole set the intent?

[da-daken]

I'll handle all exceptions inside durableExecuteAllAsync, so the caller only needs to process the outcome without additional error handling. This keeps the outer logic cleaner.

[weiqingy]

Handling it inside durableExecuteAllAsync so the caller only reads outcomes is a cleaner seam, agreed. Where would you draw the line for failures that aren't a tool's fault? A state-store write failure mapped into N per-tool outcomes would let the action complete normally, with the agent carrying on as though "all tools failed" were a real result. Would it be worth letting those keep propagating, and mapping only per-call execution failures into outcomes?

[weiqingy]

Partly. The Python catch-all is gone (tool_call_action.py:145-156), but the Java one at ToolCallAction.java:168-177 is unchanged, so the two languages now do different things with the same failure. Was leaving Java for the follow-up the intent? I answered your two-layer question in the review comment, since it bears on this.

[weiqingy]

The batch keys slots on ("tool-call-<id>", "") (tool_call_action.py:134). The Python single-call path keys on (_compute_function_id(func), args_digest) (:482-483), and since the callable is tool.call, _compute_function_id returns the same string for every tool. Java has no such split: RunnerContextImpl.java:604-605 uses getId() plus "" for the single-call state machine too.

So a job that checkpoints mid-batch and restarts with tool-call.parallel=false replays cleanly on Java, while on Python matchNextOrClearSubsequentCallResult cannot match the persisted key, truncates the journal, and calls every completed tool again. Same flip that reaches the serial-path issue on JavaRunnerContextImpl.java:193.

The per-call functionId rename was headed for its own issue, so not asking for it here. What caught my eye is that the batch path adopted it while the single-call path did not, so the two disagree inside Python today. Is documenting the flip as not recoverable across a restart the right stopgap until that lands?

[da-daken]

Yes, inconsistent IDs on the Python side would cause this issue.

Regarding functionId – I plan to address it in this PR as well. My thought is to make it exactly the same on both the Python and Java sides. What do you think? Alternatively, we can open a separate issue for further discussion if needed.

[weiqingy]

Happy either way, but I think two separable things are bundled under functionId, and only one of them is the piece we agreed to defer.

The narrow one is the divergence this PR introduces, which feels in scope here. _build_executions already mints DurableCall(id=f"tool-call-{call_id}") (tool_call_action.py:134), but _execute_sequentially passes only call.func (:174, :180), so the id is dropped and the runtime falls back to _compute_function_id → module.qualname, the same string for every tool. Java has no such split — buildCallables mints one callable and both paths use it. Threading the id that's already built through the serial path looks contained.

The broader one is using a unique per-call id as recovery identity — the "same args, different result" case we split out to its own issue. Worth flagging that this PR already crosses that line on Java: getId() returns the constant "tool-call" on main, and "tool-call-" + id here, for the serial path as well as the batch.

That carries an upgrade cost either way. An action checkpointed mid-tool-loop on the current release and restored on this build recorded functionId="tool-call", so matchNextOrClearSubsequentCallResult (RunnerContextImpl.java:654) won't match, clears from the current index, and the already-completed tool calls run again — surfaced in the log as "Non-deterministic call detected", which would point an operator at the wrong cause.

So my leaning would be to keep the serial-path id threading here as a parity fix and let the separate issue carry the semantics change. If you'd rather land both together, that works too — the piece worth writing down either way is the cross-version recovery behaviour. What's your read?

[da-daken]

Good catch — you're right that there’s still an upgrade cost here.

I agree we should split the broader semantics (“unique per-call id as recovery identity” / the same-args-different-result case) into a separate issue. For this PR I’ll revert the tool functionId back to the fixed "tool-call" on both Java and Python, so serial and batch stay consistent with the previous release and we don’t change recovery identity in this change.
Java keeps a fixed "tool-call" functionId with an empty argsDigest; Python uses qualname + args hash, aligned with serial recovery semantics.

[weiqingy]

Thanks for turning these around so fast, and the Long switch makes sense to me. Replies on the individual threads, one new thing inline, and an answer to your functionId question.

[weiqingy]

timeout_ms >= 0 makes a configured 0 mean "deadline is now" rather than "no timeout" — the first check in the loop fires and every batch fails with a TimeoutError before any tool can finish.

Java goes the other way, returning no deadline when timeout.isZero() || timeout.isNegative() (ContinuationActionExecutor.java:255), and the config table added in this PR documents it as "Non-positive disables it" (configuration.md:137). The -1 default keeps this out of the way today, but 0 is the natural thing to write for "no timeout". Should this be > 0?

[da-daken]

Thank you for the careful review～ I'll fix it in the next commit, changing it to > 0

[da-daken]

hi @weiqingy ，Thank you for the very thorough review. I have addressed the timeout configuration type, naming, and the > 0 activation behavior; kept toolActionId consistent with the previous approach; and fixed the recovery issue when switching from parallel back to serial execution.

There is still one open point regarding internal exception handling in durableExecuteAllAsync.
For the exception boundary of durableExecuteAllAsync, I prefer a two-layer model: some failures become Outcomes (caller can decide how to handle them), while others must still propagate as thrown exceptions (infrastructure / invariants).

Layer A: Should become Outcome (caller can decide)

Case	Rationale
Single callable execution failure	Already Outcome.failure; ToolCallAction can record per-tool ToolResponse
Batch timeout for unfinished slots	Already Outcome.failure(TimeoutException); partial success is meaningful
Per-slot result/exception serialization failure during finalize	Execution already finished; only persistence failed → that slot becomes failure, without affecting other successful slots’ Outcomes
Per-slot deserialization failure during replay	Treat as recovery failure for that slot; can fail and re-run without aborting the whole batch
Reconciler execution failure	Same as serial async: a per-call failure

Principle: If the error is per-call, expected, and the caller can turn it into a ToolResponse or log, wrap it in Outcome.

Layer B: Should still throw (caller must not swallow)

Case	Rationale
persistActionState() failure (Fluss / state-backend IO)	Journal is untrustworthy; continuing may break durable semantics → fail fast and let Flink retry/recover
Illegal state machine in finalizeCallAt (no slot / not PENDING / identity mismatch)	Usually a bug or serious state corruption; Outcome would hide invariant violations
reservePendingBatch argument mismatch (ids/digests size)	Programming error; runtime should not swallow it
Error (e.g. OOM)	Must not be caught and converted to Outcome
Mailbox thread check failure	Breaks the concurrency model → must throw
Durable path taken without durableExecutionContext	Configuration / lifecycle bug

I'm not sure this approach is beneficial for durableExecuteAllAsync — in that case, if ToolCallAction handles the exception here, it would mark all tool calls as failed. WDYT?

[weiqingy]

Thanks for addressing the comments. Most of the earlier items are closed, and the new java21 e2e covers the deadline leg. One item on the Python recovery path is still open, inline.

On the two-layer model

Layer B looks fully in place to me. On Layer A, three of the five rows are implemented; the two that still throw are per-slot serialization failure on finalize (JavaRunnerContextImpl.java:209-214, no per-slot catch) and per-slot deserialization on replay (RunnerContextImpl.java:527-547, reached outside any try at JavaRunnerContextImpl.java:153-155), and this PR's test pins that (JavaRunnerContextImplDurableExecuteAsyncTest.java:408-411).

On your closing question: every Layer B throw exits through ToolCallAction.java:173, which turns it into N per-tool failures and completes the action normally, so "fail fast and let Flink recover" never happens on Java and the durably-persisted SUCCEEDED slots go with it. Python honours Layer B now the wrapper is gone (tool_call_action.py:145-156). That reads to me as the model and the catch-all being mutually exclusive, independent of whether the two Layer A rows land here. How do you see the catch-all fitting once Layer B propagates?

Two smaller things:

• Layer A row 3 and the code disagree, and the new tests lock in the code: both languages abort the batch on a finalize serialization failure (test_flink_runner_context_reconcilable.py:572-608, JavaRunnerContextImplDurableExecuteAsyncTest.java:408-418), and the aborted state re-executes into the same failure on every replay. Is the table or the code the intent?
• If row 3 is adopted, persisting the slot as a failure runs through serializeDurableException (RunnerContextImpl.java:672), which can fail the same way. What would a terminal fallback there look like?

[weiqingy]

The new PENDING branch seems to open a recovery gap on Python, and I would welcome a sanity check on the routing. matchNextOrClearSubsequentCallResult returns null for a matching PENDING slot without clearing it or advancing currentCallIndex (RunnerContextImpl.java:762-771). Java pre-checks PENDING in durableExecuteCompletionOnly at :297-300 and finalizes in place, so it never lands there. Python's _try_get_cached_result reads that null as a plain miss (:494), re-executes, and recordCallCompletion appends (RunnerContextImpl.java:813): a two-call batch replayed serially becomes [PENDING(f0), PENDING(f1), SUCCEEDED(f0), SUCCEEDED(f1)], and each later failover re-runs both tools and appends two more.

It is new at this head: at 9169bb2 the batch reserved "tool-call-<call_id>", which never matched the serial digest, so the mismatch branch cleared and re-ran cleanly. No config flip is needed either, since len(executions) dropping to 1 on an unresolvable tool routes a reserved batch to the serial path (tool_call_action.py:69, :108-117); flipping back to parallel caps the growth but leaves a stale tail. Java has the recovery test (RunnerContextImplDurableExecuteTest.java:267-299); would a Python counterpart make sense, given the parity ask in AGENTS.md? For the fix, the bridge could take Java's pre-check, or the matcher could clear the slot before returning the miss. Which fits the bridge contract better?

[weiqingy]

The fake's matchNextOrClearSubsequentCallResult (:103-116) has no status == "PENDING" branch, so a PENDING match still returns [True, None, None], the pre-fix behaviour changed at RunnerContextImpl.java:762-771. A Python regression test for the batch to serial case would pass here while production appends. Would it help to teach the fake the new contract first, so such a test could fail?

[weiqingy]

id is required here but the Python runtime never reads it: _durable_identity (flink_runner_context.py:751-752) recomputes identity from call.func/args/kwargs at :755, :821, :843. Java treats the id as the journal identity (RunnerContextImpl.java:293), so a stable id with varying arguments keeps recovery on Java and loses it on Python. DurableCall is new, so it is cheap to settle now. Should id stay on the dataclass at all?

Keeping it also pulls in the plan to runtime import (tool_call_action.py:34), the first production one in the repo, against the direction AGENTS.md states. Java declares the id locally (ToolCallAction.java:45). Where would you want that to live?

[weiqingy]

The code side is resolved: DurableCall.id is gone, and tool_call_action.py no longer imports from flink_agents.runtime.

On the tech-debt follow-up that was suggested for bringing Python's durable identity back in line with Java's getId() contract, I could not find an issue tracking it. The closest is #956, which is scoped to the Tool result contract (ToolResponse vs Any) and does not touch recovery identity or the class names. Would it be worth opening one before this merges? Deferrals without an issue behind them tend to quietly disappear.

[weiqingy]

durable_id is on both public abstract signatures (:244, :306) and threaded through flink_runner_context.py, but nothing passes it, and test_tool_call_action.py:361 asserts it is not passed. Java has no counterpart. Two things that made me pause while it sits unused: reconciler's docstring marks itself reserved and not forwarded to func (:285-286) where durable_id's (:287-290) does not, so a tool declaring its own durable_id would have it swallowed, and every third-party RunnerContext inherits the abstract signature. Is a follow-up expected to use it?

[weiqingy]

nit: two JDK wording points. The new row has no JDK caveat, though the option is inert below Java 21: the java11 fallback takes timeout and ignores it (ContinuationActionExecutor.java:75-87), and two of the five IT combos run Java 17. Conversely :135 says tool-call.parallel needs JDK 21+, which does not hold for Python, whose batch runs on a ThreadPoolExecutor (flink_runner_context.py:266). Worth a clause on each?

[pltbkd]

Thanks @da-daken for proposing this. The thorough code review from @weiqingy already covers a lot — thanks for that. On top of it, I'd like to raise a few concerns of my own.

1. DurableCall vs Java's DurableCallable — not aligned

The new Python DurableCall and Java's DurableCallable are not aligned: Java uses the caller-declared getId() for recovery matching, while Python never consumes any caller-supplied identity and derives it from func + args instead. This seems to conflict with the requirement that new APIs stay semantically aligned across languages (and the class names differ too).

That said, this is essentially a legacy gap, and this PR actually partly closes it. From a semantic-alignment standpoint it's fine to keep the current state (including the removal of the id field) for v1; I'd just suggest we track it as a tech-debt follow-up to bring Python's durable identity back in line with Java's getId() contract — including aligning the class names.

2. Timeout — confirm reconcile impact before merge; and subagents will need this too

The timeout here is a mechanism on durable_execute, but the framework has no way to actually cancel a running callable. If the callable has side effects, then after a timeout the framework records the state as failed while the side effects may still happen — leaving flink-agents state inconsistent with the actual side effects. The exception path is also long, and the exception type is lost on recovery replay (TimeoutException becomes a bare RuntimeException).

I'm not asking to remove the timeout here. But before merging, I'd like to confirm the interaction with reconciler: reconciler only fires on PENDING slots, and timeout finalizes the slot to FAILED — so after a timeout, the reconcile channel looks closed. Is that the intended behavior?

Also worth noting: subagents are expected to use the durableExecuteAllAsync API too. If the mechanism can't properly support side-effecting callables (cancel + post-timeout state semantics), that gap will need to be filled soon — not just for tool calls.

3. Do we really need two thread pools?

The motivation in the discussion for a dedicated pool was to avoid one ToolRequestEvent exhausting the global pool and affecting other keys. But if every record carries multiple parallel tool calls (which is the typical case), the tool pool gets saturated just the same — the problem simply moves to the other pool. Meanwhile, the default thread count is based on CPU cores, a sizing heuristic meant for compute-bound work; tool calls are I/O bound, so adding another pool of the same size only increases CPU and memory pressure.

I think max-parallelism (a single shared pool + a cap on how many threads one record/batch can occupy) may be more suitable. It can be a follow-up optimization. But if that's the direction, I'd suggest avoiding introducing a separate tool-call.num-async-threads config now, since it would likely need to be removed or reworked when we move to that model.

4. Config surface — potential redundancy given the above

Following from point 3: if max-parallelism turns out to be the more suitable direction, tool-call.num-async-threads looks redundant — the single pool is already sized by num-async-threads. At the same time, the new tool-call.parallel boolean switch could simply be folded into parallelism = 1 (serial behavior). So the current three-option surface looks like it carries config we'd likely revisit or collapse shortly.

[da-daken]

Thanks @weiqingy @pltbkd for the new review — every point was helpful.

I’ve removed the extra id parameter on the Python side. Whether Java and Python should use the same recovery identity is worth discussing in a follow-up issue, as @pltbkd noted in the first point. I’ve also aligned Python recovery behavior with Java: when a PENDING slot is read, we re-execute and recover in that same slot — that was my oversight, thanks for catching it.

The durableExecuteAllAsync-related changes haven’t been pushed yet — sorry for the confusion. The catch-all semantics I mentioned earlier no longer apply. What I meant was that system/infrastructure exceptions should propagate so users can see them; in ToolCallAction, we can keep the same approach as the serial path — a system failure surfaces as tool-level failures for all affected tools.

On Layer A serialization failures: I’d like to correct my earlier wording — we should return a failed Outcome without persisting the slot as failed; the slot stays PENDING. On JDK < 21, we also don’t persist the slot; we throw instead, so on recovery we run call/reconcile, which matches the intent. When a system error happens before the slot is persisted, running call/reconcile on recovery is also what ToolCallAction expects.

Re @pltbkd’s review: durableExecuteAllAsync already has a real use case — execution timeout. From the framework’s perspective that’s a failure; we surface a timeout exception so callers can retry or compensate (which would land in another slot). That made the old toException() behavior — wrapping everything in RuntimeException — more painful, so I extended it to restore the original exception type after deserialization.

On the two thread pools question: I agree with a single shared pool + a cap on how many threads one record/batch can occupy, with batch parallelism = cores. A second pool would use extra memory without real benefit over enlarging the shared pool; semantically, a dedicated pool for durableExecuteAllAsync isn’t necessary — one pool is enough. Because parallel tool execution is on by default, it can contend with other keys; I’ll call that out clearly in the ConfigOption docs so users notice the behavior change.

On configuration: agreed — we can simplify to a single knob: parallelism = 1 (serial) vs parallelism > 1 (parallel).

[weiqingy]

Thanks for addressing the comments. The Python recovery alignment landed, and the new reconcilable tests really do bite: run against the previous code they fail, so they are genuine regression tests rather than assertions that happen to pass. One heads-up before the inline comments: CI is red at head, with it-java [java-21] cancelled at the 40-minute timeout on all three Flink combos.

[weiqingy]

I think this barrier can never be completed. batchBarrier is created here and handed to setPendingBatchFuture, but the only thing that completes it is :235, which runs after the while (completed < batchSize) loop finishes. The loop yields at :231 and needs executeAction to resume it, and executeAction:72-74 will not resume while hasPendingAsync() is true. ContinuationContext.java:74-76 reports true for exactly as long as the barrier is pending. So after the first yield the action stops making progress. No thread blocks, the task just re-queues on the mailbox forever (ActionExecutionOperator.java:424).

With the default tool-call.batch.timeout.ms = -1 (AgentExecutionOptions.java:81-82) there is no deadline to fall back on, since getDeadlineNanos:295-299 returns Long.MAX_VALUE. Setting a timeout does not really rescue it either: the resume then lands straight in the timeout branch at :200, so the success path at :235 is effectively unreachable.

This is on by default now, because tool-call.parallelism defaults to availableProcessors() (:69-73) and ToolCallAction.java:69 sends any two-tool batch down the parallel path. CI seems to agree: it-java [java-21] is cancelled at the 40-minute timeout on all three Flink combos at head, green at 5c9a67e3a and red from b293e5cd6 on. In job 92813859324 the test that stalls is ReActAgentTest, which is already on main, so it is not just the new tests. The two java-17 combos take the serial fallback and pass in about 10 minutes.

The old CompletableFuture.allOf(...) let the pool complete the barrier from outside, which is what made the gate work. Would going back to that shape, and using the window only to decide when to submit, be enough to fix it?

[weiqingy]

When the deadline hits, slots the window never reached are still null, so each one becomes Outcome.failure(timeoutException) here. JavaRunnerContextImpl.finalizeExecutedOutcomes:185-201 then persists them through finalizeCallAt as Status.FAILED. With parallelism=2, a 10-tool batch that times out records eight tools that never started as permanently failed, and they come back as cached failures on replay. Python does the same at flink_runner_context.py:379-381.

This is separate from the resume problem above, since it sits on the timeout path that a fixed barrier would still take. Before the sliding window every supplier was submitted up front, so at least every tool had started. It is also close to the open question about timeout and the reconciler, though a slot that never started feels like a different case. Would leaving those slots PENDING, so recovery can still run them, fit better than recording them as failed?

[weiqingy]

Changing Optional.of(null) to Optional.empty() fixes a real NPE, but it also makes a cached null look exactly like a cache miss, and both callers only check isPresent().

On the completion-only path (:302-316) that means the call runs a second time, and recordCallCompletion:827-846 appends another CallResult and advances currentCallIndex again after the hit already advanced it at :799, so every later slot in the action shifts by one. On the reconcile path (:649-659) it throws IllegalStateException saying the slot is not terminal, when it is.

Any durable call that legitimately returns null lands here, since serializeDurableResult returns null (:687-692) and new CallResult(fid, digest, null, null) counts as SUCCEEDED (actionstate/CallResult.java:93-101). Tool calls are safe because ToolCallAction always returns a ToolResponse, so this is really about user code calling durableExecute. Same shape as the Python miss you just fixed. Would a separate hit flag, or a sentinel for the absent slot, be enough to tell the two apart?

[weiqingy]

This is the toException() work you mentioned.

The no-arg fallback here drops message, so the rebuilt exception has getMessage() == null, and ToolCallAction.recordExecutionException:236 then writes a null into the ToolResponseEvent error map. The old wrapper always carried exceptionClass + ": " + message. Could the message be carried through on this path too?

Less certain, and I have not tested it: Class.forName(exceptionClass) at :359 is the one-arg form, so it resolves the class with RunnerContextImpl's own loader rather than the user-code loader that JavaActionTask:67 installs. Framework types like TimeoutException resolve either way, so whether this bites probably depends on where the flink-agents jars sit.

[weiqingy]

cancel(true) does not interrupt anything on a CompletableFuture, since mayInterruptIfRunning is ignored there, so a tool that hangs keeps its thread until it returns on its own. Python has the same limit (flink_runner_context.py:383), so this looks like a shared gap rather than a Java one.

What changed this round is which pool that thread comes from. With the dedicated tool pool gone, JavaRunnerContextImpl.executeOutcomeSuppliers:244 submits to the same continuationExecutor as chat and RAG, so one hung tool now costs every key on the subtask a thread. The javadoc at AgentExecutionOptions.java:57-67 covers the sizing side of this. The part it does not cover is that after a batch timeout the thread never comes back. Is there a way to bound that, or is documenting it the right call for v1?

[weiqingy]

This assertion is failing at head: ut-python [macos-latest] [java-17] [python-3.12] reports assert 0.40510524999990594 < (0.2 * 2). A 2x margin on a 200 ms sleep leaves roughly 200 ms for thread start-up on a shared runner, and the suite has gone red on a different Python version on each of the last three commits. Would a wider margin, or checking that the intervals overlap instead of total wall-clock, hold up better here?