GitHub user da-daken added a comment to the discussion: Parallel Tool Call
Execution
hi @weiqingy @joeyutong Thanks for the sharp review — all three points were
spot on. Here's how I've addressed each in the updated design.
---
### 1. Python parity
You're right that `asyncio.gather` won't work and that `__await__` currently
fuses execute + record.
First, the three structural constraints on the Python side:
1. **Index state lives entirely in Java** — Python has no
`DurableExecutionContext`, it only bridges call-index reads/writes through
`_j_runner_context`, and today's bridge is "current index" only.
2. **`__await__` fuses execute + record** —
`_DurableAsyncExecutionResult.__await__` calls `_record_call_completion` right
after the thread pool finishes and does `currentCallIndex += 1`.
3. **No asyncio event loop, one `send(None)` per operator tick** — multiple
independent `await`s can only advance serially, so you can't get concurrency
within a single action.
What I changed:
- **First-class batch entry**: `ctx.durable_execute_all_async([...])` returns a
single awaitable that does the fan-out internally — sidesteps `gather`'s "one
await = one slot" assumption.
- **Unfuse execute from record**: a new
`_DurableBatchAsyncExecutionResult.__await__` does submit-all → a single `while
...: yield` loop over all futures → record in tool_call order. Recording no
longer hangs off a single await; it's written at absolute indices when the
batch settles.
- **Index-addressable primitives (Java side, shared by Java + Python)**:
`reservePendingBatch` (reserve N PENDING at once),
`getCallResultFieldsAt(index)`, `finalizeCallAt(index, ...)`,
`advanceCallIndexBy(n)` on `DurableExecutionContext`. This is exactly the
missing "absolute index" you called out — today's `appendPendingCall` only
appends one at the current index (a second consecutive call throws), so
`_record_call_completion` stops being the only recording path.
- **Aligned data model**: Python gets `DurableCall` (the `DurableCallable`
equivalent, with `id` / `reconciler` and a reserved per-call `timeout`) and
`Outcome` (three states, `.value` / `.error`); each tool uses a stable id
`f"tool-call:{call_id}"` instead of every tool sharing the single
`FunctionTool.call` id.
- **Recovery & reconciler**: the prepare phase scans the cache by absolute
index (hits skip submission; `PENDING` with a reconciler goes through the
reconciler) — same semantics as the Java three-phase flow.
### 2. Timeout fan-in
Good catch — the batch would otherwise hang on a stuck tool. Now:
- On the batch deadline, every unfinished slot is finalized via
`finalizeCallAt(..., BatchTimeoutException)` so fan-in can always collect N
results — no dangling `PENDING`, deterministic recovery.
- I kept per-call deadlines (v1 uses one batch value; per-tool timeout later
just changes how the deadlines are computed — no strategy class needed).
- On the `cancel(true)` distinction: documented explicitly. It bounds the
*wait*, not the *thread* — a tool that ignores interruption keeps its pool
thread until it returns, temporarily shrinking `num-async-threads`. Guidance
added: framework timeout is a backstop; tools that can block should use
interruptible / self-timing clients. (Python is even blunter here —
`Future.cancel()` can't stop a running thread at all, so that's flagged too.)
### 3. Return type
Fully agree — `List<T>` can't tell a thrown tool from a `null` result without
reading slot state back. The batch API now returns per-call `Outcome<T>`:
- `SUCCEEDED` / `FAILED` / `TIMED_OUT`, with `value()` / `error()`.
`success(null)` and `failure(e)` are distinguished by status, not by `null`.
- This lines up 1:1 with `ToolCallAction`'s `success` / `error` / `responses`
maps — no more round-tripping through the store.
- Timeout stays faithful across recovery: it's persisted as `FAILED` carrying
`BatchTimeoutException`, and rebuilt back into `TIMED_OUT` by exception type,
so metrics don't lose it.
Roughly:
```java
public final class Outcome<T> {
public enum Status { SUCCEEDED, FAILED, TIMED_OUT }
public static <T> Outcome<T> success(T value); // value may be null
public static <T> Outcome<T> failure(Throwable e);
public static <T> Outcome<T> timeout(BatchTimeoutException e);
public boolean isSuccess();
public boolean isFailure();
public boolean isTimeout();
public T value(); // success value (may be null)
public Throwable error(); // failure / timeout exception
}
// Callers branch on status instead of guessing from null:
Outcome<ToolResponse> o = outcomes.get(i);
if (o.isSuccess()) {
responses.put(id, o.value());
} else { // FAILED or TIMED_OUT
error.put(id, describe(o.error()));
}
```
---
One extra thing that fell out of this: missing tools now occupy a response slot
with a pre-set `FAILED` outcome (never submitted to the pool) so indices stay
strictly aligned.
GitHub link:
https://github.com/apache/flink-agents/discussions/855#discussioncomment-17622632
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
