[runtime][java] Avoid async-pool self-deadlock in internal sub-agent await (#1213)

[yunfengzhou-hub]

Linked issue: close #1213

Purpose of change

A Java action awaiting an internal sub-agent no longer occupies an async-pool worker, so a child making its own durable async calls completes at num-async-threads=1. Previously the wait blocked a worker while the child's calls queued for one from the same pool; neither advanced.

Runtime flow

On resolve, DeferredSubagentFuture.execute() calls durableExecuteAsync(prepare()). prepare() already runs on the mailbox thread — registering the call status, sending the child's envelope, failing fast without stackful suspension — and now returns a ContinuationNativeCallable.

JavaRunnerContextImpl.executionBoundary is the only decision point added: callable::call for a marked callable, resolveAsync(callable::call) otherwise, used by resolveDurableAsync on both the completion-only and the reconcile path. A marked callable therefore runs inline in the caller's continuation on the mailbox thread and hands the response future to ContinuationActionExecutor.awaitFuture. That registers future.thenApply(v -> null) as pendingFuture and yields until it completes; executeAction's existing hasPendingAsync gate resumes it on the mail loop. No worker is held, so the child draws on the whole pool, and the callable returns the real SubagentResult.

Key decisions

• A marker interface on the callable carries the fact its producer alone knows — that the body yields the caller's continuation — to the dispatcher.
• Yield rather than a dedicated thread or a bigger pool: another thread moves the exhaustion elsewhere, and more workers only raise the threshold — nesting and concurrency close the cycle at any size.
• Keep the public blocking awaitSubagentCall: Python calls it on its own worker, where blocking is correct.

Behavioral Semantics

Interaction decisions

Callable	Caller	Await shape	Behavior
Marked	Java, JDK 21+	single await()	Inline; the wait yields; no worker held
Marked	Java, JDK 21+	combined awaitAll()	Each handle crosses the boundary in turn
Marked	Java, JDK 21+	nested	Every level yields; depth consumes no workers
Marked	Java, JDK 21+	response future already complete	Loop guard false on entry; never suspends
Marked	Java, ≤ JDK 17	any	requireMailboxSuspension raises first; unchanged
Marked	Python async action	any	Never reaches prepare; blocks in awaitSubagentCall
Unmarked	any	any	Unchanged: dispatched to a pool worker
Any	any	result already in durable state	Cached value returned; neither boundary reached
Marked + reconciler()	any	any	Same inline boundary; no producer supplies one today

Behavioral contracts

1. Awaiting an internal sub-agent holds no async-pool worker, so a child making its own durable calls completes at num-async-threads=1.
2. Nested calls, and calls combined with combine(...) and awaited by awaitAll(), complete at num-async-threads=1.
3. A successful call still yields the child's accumulated output; a child failure still surfaces as a failed SubagentResult carrying the child's exception.
4. A completed call is still durable: the parent's recorded result is the child's real output.
5. An unmarked DurableCallable still runs on a pool worker, unchanged.
6. A Java caller without stackful suspension still fails fast rather than blocking the mailbox.

Failure behavior

• Child action raises: the response future completes exceptionally, call() unwraps the cause and folds it into SubagentResult.error; the job proceeds.
• An Error cause is rethrown as-is rather than folded into a result; a CompletionException with no cause is rethrown itself.
• Await for an unregistered (sessionId, callId), and resolve without stackful suspension, each raise IllegalStateException; unchanged.

Tests

Contract	Tests
1. No worker held	singleAsyncThreadDoesNotDeadlockWhenChildUsesDurableCall
2. Nesting	singleAsyncThreadDoesNotDeadlockOnNestedCalls
2. Combined calls	singleAsyncThreadDoesNotDeadlockOnCombinedCalls
3. Success and failure results unchanged	callerReceivesTheChildsAccumulatedOutput, childFailureSurfacesThroughTheResult
4. Durability across replay	InternalSubagentRecoveryTest, 14 cases
5. Unmarked callables unchanged	existing durable-execution suites
6. Fail-fast without continuations	resolvingWithoutContinuationsFailsFast
Marked + reconciler() runs inline	testDurableExecuteAsyncContinuationNativeReconcilableRunsInline

The three deadlock tests are @Timeout(30) at num-async-threads=1, gated to JDK 21+, pinning the pool size where the cycle closes and the two shapes that multiply waits in flight. Full runtime suite green on both JDKs.

This PR also fixes the ut-python failure its first push hit: test_flink_runner_context_gather_timeout_keeps_completed_results (#1208; untouched by the Java diff) raced thread startup against a 100ms wall-clock deadline, so a loaded runner persisted the timed-out slot as PENDING, not FAILED. The four tests asserting a slot started first now wait for their workers to say so.

Not verified: live-cluster failover of an in-flight call under the inline boundary (recovery cases are harness-level); pool sizes above one.

Rejected alternatives, mutation notes, and implementation invariants

Alternatives rejected. A default method on DurableCallable would let a user mark their own callable, and inlining one onto the mailbox pins it for the whole action. Matching on InternalSubagentSetup would tie the generic context to one implementation and encode identity rather than the requirement. Bridging through a new resumption path would duplicate executeAction's gate and the operator's mail loop.

Mutation notes. Dispatching the marked callable to the pool as before makes all three deadlock tests time out at 30s each rather than pass quietly. Restoring the reconcile path's hard-coded resolveAsync supplier makes testDurableExecuteAsyncContinuationNativeReconcilableRunsInline fail expected: <0> but was: <1> on the pool-dispatch count.

Implementation invariants, not caller-observable:

• awaitFuture leaves a done bridge in pendingFuture when the response future was already complete on entry. hasPendingAsync() tests isDone(), so it does not read as pending, and executeAction clears the slot before resuming. executeAsync leaves the same shape.
• The bridge exists only so the existing Future<?> slot can carry a CompletableFuture; nothing reads its value or exception. getPendingFuture() has one reader, which nulls it without joining.
• Resumption is polled, not signalled: no completion callback exists in the operator package, so a bridge already done on entry cannot strand the continuation.
• Continuation.yield requires the calling thread to have entered the scope, which only the mailbox thread does inside executeAction. That is why the boundary switch is a correctness requirement, not an optimization.
• A combined awaitAll() reaches the boundary once per handle: SubagentFutureGroup prepares every handle, then executes them one at a time through execute().
• executionBoundary now governs both durable async paths: resolveDurableAsync computes it once and passes it to durableExecuteWithReconcile as well as durableExecuteCompletionOnly. Before this, the reconcile path built its own resolveAsync supplier, so a marked callable that also supplied a reconciler() would have yielded on a pool worker and thrown NoSuchElementException. No producer supplies both today, so the change is behavior-neutral; the new unit test pins the routing.
• Signatures verified untouched: DurableCallable, SubagentSetup, SubagentFuture, RunnerContext, both awaitSubagentCall entry points.
• Python flake evidence: with every positive async.batch.timeout.ms in the file shrunk to 1, the pre-fix tests reproduce the CI assertion ('PENDING' == 'FAILED') and the fixed ones still pass, so the fix removes the race rather than widening the window. The four tests are driven by a new _run_async_after_workers_started helper that runs the batch generator one step, waits for the workers' own events, then drains it; the deadline has certainly passed by then. async.batch.parallelism is now pinned in the first of them so submission of both children does not depend on the runner's core count.
• A narrow production race remains in the same Python code and is out of scope here: if the deadline trips after the pool dequeues a task but before _mark_started_on_run writes its flag, the slot is persisted PENDING while the task actually runs, so recovery can duplicate its side effects.
• The JDK 11 awaitFuture blocks on future.get() for multi-release compilation only, and is unreachable here because requireMailboxSuspension rejects first.
• Change size: 10 files, +516/−55 — 9 under runtime, 1 Python test file.
• Suite counts: runtime green on JDK 11 (1240 tests, 19 skipped) and JDK 21 (1244 tests, 27 skipped); the reconcilable Python test file green at 52 tests.

API

No public API change: ContinuationNativeCallable lives in runtime, not api, and is not part of any published contract. Every existing signature, the durable state format, and the Python path are untouched.

One thing changes: num-async-threads no longer has to exceed the in-flight internal sub-agent waits. JDK 21 is still required.

Documentation

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

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

• Yes
• No

Generated-by: Qoder (Qwen 3.7 Max)

[wenjin272]

Thanks for the fix! The continuation-based approach addresses the shared-pool deadlock on JDK 21+, and I have no further concerns with this PR.

We also plan to support Java internal sub-agent calls on JDK 11/17, where this continuation mechanism is unavailable. That will need a separate design for suspending the caller without blocking the mailbox or occupying an async worker. We can follow up on that separately; it does not block this PR. cc @pltbkd