wenjin272 opened a new pull request, #1127: URL: https://github.com/apache/flink-agents/pull/1127
Linked issue: #1122 ### Purpose of change This makes asynchronous durable calls composable in Java and Python without requiring users to construct a separate batch-call descriptor. #### Runtime flow 1. `durableExecuteAsync` / `durable_execute_async` creates a cold, runtime-owned `DurableFuture`; creation neither reserves a durable slot nor starts the callable. 2. Resolving a single handle through Java `ctx.await(...)` or Python `await` delegates to the existing recovery state machine, which replays a terminal result, reconciles or re-executes a pending call, or reserves and executes a new call. 3. Resolving `ctx.gather(...)` plans the complete ordered batch, reserves all required pending slots, and only then submits uncached calls. 4. Gathered outcomes complete the child handles in input order, so a later await replays the locally cached value or exception without consuming another durable slot. #### Key decisions - `DurableFuture` is an opaque, context-owned handle rather than a `java.util.concurrent.Future`. Java waits go through `ctx.await()` so JDK 21 execution can yield through the runtime continuation instead of blocking with `Future.get()`. - Both language APIs remain deferred. This preserves the reserve-before-execute invariant; eager submission would require a separate reservation protocol. - `gather` accepts only unique, unresolved single-call handles created by the same context. This prevents ambiguous slot ownership and reuse. - The old public batch descriptor and batch execution APIs are removed; their recovery and parallel execution machinery remains internal. - Durable identity matching is unchanged. The broader identity model remains tracked by #1016. ### Behavioral Semantics | Interaction | State | Behavior | | --- | --- | --- | | Await one handle | New call | Reserve, execute through the async runtime path, persist, and cache locally. | | Await one handle | Terminal or pending recovery state | Replay the terminal outcome, or reconcile/re-execute the pending call. | | Await a gathered handle | Mixed cached, pending, and new calls | Preserve input order; reserve every required new slot before starting any uncached call. | | Await a resolved child | Locally completed by `await` or `gather` | Return or rethrow the cached outcome without advancing durable state. | Behavioral contracts: - Creating or abandoning a handle has no execution side effect; this change does not add action-end validation for unresolved handles. - Single-call failure is rethrown. Gathered callable failures are returned as per-call `Outcome.failure` values. - `InterruptedException` remains a cancellation signal in Java: it is propagated, the interrupt flag is restored, and interrupted/unstarted slots remain pending. - On batch timeout, started calls are finalized according to their outcomes while unstarted calls remain pending for recovery. - A serialization/finalization failure is returned as a failure outcome and leaves that slot pending. - JDK 21 uses the continuation-backed async path; older JDKs retain the synchronous fallback. Python handles remain directly awaitable but do not become general-purpose `asyncio` futures. ### Tests | Contract | Coverage | | --- | --- | | Cold and reusable handles | `testDurableExecuteAsyncCreatesDeferredReusableHandle`; `test_flink_runner_context_async_future_is_deferred_and_reusable` | | Reserve before execute, ordering, and child completion | `testGatherReservesBatchBeforeExecutionAndCompletesChildren`; `test_flink_runner_context_gather_reserves_before_execution` | | Replay, reconciliation, and partial recovery | Java and Python gather recovery/reconciler suites in the runtime context tests | | Failure, timeout, interruption, and pending-slot behavior | Java gather timeout/interruption/finalization tests and Python gather failure/timeout/finalization tests | | Production tool-call migration | Java and Python `ToolCallAction` tests, including parallel business and infrastructure failures | Verified: - Java plan tests: 52 passed. - Java runtime/context compatibility tests: 44 passed. - Python non-integration suite: 1347 passed, 13 skipped, 172 deselected. - `./tools/lint.sh --check`. - `./tools/build.sh --java` on JDK 17: all 35 modules built successfully with tests skipped. - JDK 21 runtime multi-release and test-source compilation. Not verified: external-service E2E execution, live Flink checkpoint recovery, and execution of the JDK 21 continuation path (compile-only in this environment). <details> <summary>Implementation invariants and evidence</summary> - Runtime-only future implementations keep callables and internal batch plans out of the public API. - Java and Python validate owner, handle kind, duplicate inputs, and already-resolved inputs before constructing a gathered handle. - Existing tool, chat-model, context-retrieval, subagent, memory, E2E, and documentation call sites were migrated to direct await or gather composition. </details> ### API This is a breaking pre-0.4 Java API change: `durableExecuteAsync` now returns `DurableFuture<T>`, and immediate callers migrate to `ctx.await(ctx.durableExecuteAsync(...))`. Java adds `await` and `gather`, and removes `durableExecuteAllAsync`. Python direct-await usage remains source-compatible. Python adds public `DurableFuture` and `ctx.gather(...)`, removes public `DurableCall` / `AsyncExecutionResult`, and removes `durable_execute_all_async` in favor of composing single-call handles. ### Documentation - [ ] `doc-needed` - [ ] `doc-not-needed` - [x] `doc-included` ### Was this patch authored or co-authored using generative AI tooling? - [x] Yes - [ ] No Generated-by: Codex (GPT-5) -- 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]
