[api][runtime][java][python] Add composable durable futures

[wenjin272]

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 [Tech Debt] Align Java and Python recovery model #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).

Implementation invariants and evidence

• 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.

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
• doc-included

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No

Generated-by: Codex (GPT-5)

[wenjin272]

@pltbkd @da-daken @joeyutong Would you mind taking a look at this PR at your convenience?

[da-daken]

LGTM! Just a non-blocking comment—take a look when you get a chance.

[da-daken]

Would it be better to add a description about parallelism?