[dangzitou]

Linked issue: #1140

This draft is stacked on #1138 and includes its internal sub-agent framework prerequisite. The batch-specific change is 9ad3a7b1.

Purpose of change

Parents can combine internal sub-agent handles into a durable batch, run a bounded fan-out across the same or different children, and receive ordered per-call results. This removes serial resolution and shared-worker starvation from the grouped invocation path.

Runtime flow

submit allocates deterministic identities without starting internal children. Group resolution validates admission, prepares deferred handles, and gathers their durable futures. The runtime reserves every outstanding slot, starts a bounded window on the mailbox, suspends the parent, persists each terminal result before admitting its replacement, and returns results in handle order. Recovery replays the parent first to reconstruct child scopes and reuses child action journals.

Key decisions

• Reuse the existing group API and durable batch machinery; internal completion uses futures/coroutine polling instead of blocking async workers.
• Bound active invocations per gathered batch, including different resource names. Nested batches have independent capacity so waiting ancestors cannot starve descendants.
• Bound queued state by rejecting an oversized group before reservation; queued calls use existing PENDING slots rather than a second journal format.

Behavioral Semantics

Interaction decisions

Conditions	Behavior
Internal or mixed internal/deferred group	One sub-agent window; tool-call parallelism and timeout do not apply.
External deferred group only	Existing tool-batch scheduling options apply.
Already submitted async handles	Await existing runs; grouping cannot retroactively control their submission.
Completed and outstanding handles	Reuse completed results; schedule outstanding deferred calls in their original order.
Nested internal group	Its own window; the ancestor occupies only its parent's batch slot.
Child failure with unfinished actions/events	Retain the slot until the child scope quiesces, then return its failed result.

Behavioral contracts

1. A group returns one result per handle, in input order, after all child calls settle; successful siblings survive child failures.
2. Internal groups admit at most subagent.parallelism invocations at once and work with one shared async worker, including nested calls.
3. Oversized groups, duplicate handles, different owning contexts, and non-positive limits fail before internal preparation/reservation.
4. Recovery preserves terminal results and replays running/queued calls under their original identities, including intermediate child events and failure summaries.
5. Cancellation and infrastructure errors propagate from resolution; infrastructure errors leave unfinished durable slots pending.

Failure behavior

Admission errors raise. Java internal calls require JDK 21+ continuation support; Python callers must be async. Child action failures become SubagentResult.error after quiescence and their summaries survive replay. Bootstrap, polling, serialization, and journal errors propagate rather than becoming child results. Unfinished external effects may replay under the existing durability model, so callers still need idempotency or a reconciler.

Tests

Contract	Tests
Ordered results and partial failures	InternalSubagentBatchTest, Python test_internal_subagent_batch, Java/Python batch E2E
Bounded concurrency, one worker, nesting	Java/Python batch E2E; existing nested-call tests with parallelism/worker count 1
Admission without side effects	Invalid-group cases in Java/Python batch unit tests
Terminal/running/queued recovery	InternalSubagentRecoveryTest, Java checkpoint/restart E2E, Python partial-result replay test
Exceptions versus durable child results	Java bootstrap-failure/quiescence tests; Python bootstrap/polling-failure tests; existing cancellation tests

Coverage targets scheduling, worker starvation, durable slot indexing, checkpoint queue rotation, nested event replay, and the interval between child failure and parent result persistence.

Not verified: batch recovery against a production Kafka/Fluss journal, process/TaskManager loss, or rescaling across TaskManagers. The Python MiniCluster restart verification is local/manual and is not a committed CI regression.

Implementation invariants and verification evidence

All slots are reserved before bootstrap. Admission counts submitted slots until their completion callback persists them. Recovery drops checkpointed child tasks and regenerates their graph from the waiting parent, avoiding dispatch before transient scopes exist. Completed actions replay same-call forwarded events and outputs, but not nested bootstrap commands. Failed actions persist their failure summary and release discarded event counts. Nested AgentPlan JSON now closes its root object explicitly so resource-provider type markers remain in the correct object.

Core verification uses JDK 21, Python 3.12 and Flink 2.3. Distribution artifacts are rebuilt and copied to the Python library directories as in tools/build.sh.

• mvn -B --no-transfer-progress -pl runtime -am -Dspotless.skip=true test: 2,190 tests, 39 skipped, no failures.
• ./tools/ut.sh -p: 1,670 passed, 14 skipped; final core regression: 1,289 passed, 14 skipped.
• Java InternalSubagentBatchE2ETest: both cases pass locally on Flink 1.20, 2.0 and 2.3 with JDK 21 and in CI on Flink 2.0/2.1/2.2; older Flink uses java.util/java.nio module opens. Fixtures use shared source/collector APIs. Java 17 + Flink 1.20 discovers both JRE-gated cases and passes 5 existing integration/serialization cases on three consecutive runs.
• Python internal_subagent_test.py: single call, failure, nested call and bounded 12-call mixed-target batch.
• Local Python-parent MiniCluster restart fixture on Flink 2.3.0, JDK 21.0.12.1 and Python 3.12.14: 6/6 runs (three successes and three cached child failures). Failure follows a completed checkpoint with one terminal, two running and one queued call. The agent operator restores, original call identities match, child invocation counts are [1,2,2,1], and ordered results/failures survive. The journal uses production ActionState serialization in the same JVM; a test-only close hook releases parked Python work during task shutdown. This fixture is uncommitted.
• ./tools/lint.sh -c with JDK 17; ./tools/check-license.sh; tools/check-agents-md.py.

API

No invocation signatures change. Combining deferred handles now resolves one durable batch; sequential waits remain sequential. New Java/Python/YAML configuration keys are subagent.parallelism (16) and subagent.max-batch-size (1024), both positive. These bound call counts per group, not payload bytes or total job concurrency. Single internal waits use the same scheduler. External async submission remains unchanged.

Documentation

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

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

• Yes
• No

Generated-by: Codex CLI 0.154.0 (GPT-6)

[yunfengzhou-hub]

Hi @dangzitou , #1138 has been merged, and some prerequisites this PR bases on might have been changed. Could you please rebase your PR on the latest code of main branch and resolve the conflicts?

[dangzitou]

Hi @yunfengzhou-hub, thanks for the heads-up.

After rebasing, I see that #1138 landed with a different internal-call model, and #1220 then moved internal sub-agent execution onto SubgraphInvocation (removing InternalSubagentCallStatus). So this PR can't be rebased mechanically: the admission and queueing logic needs to be rebuilt on the new invocation model.

Instead of one ~4.9k-line PR, I'd like to split #1140 into smaller PRs:

1. Bounded admission for internal sub-agent invocations in the Java runtime: a concurrency option plus queueing/backpressure on top of SubgraphInvocation, with unit tests.
2. Durable state for queued-but-not-started calls, and recovery of an in-flight batch.
3. Python runtime parity.
4. E2E tests and the Sub-Agents guide.

Two questions before I start:

• Should the concurrency bound apply only to internal sub-agent invocations, or to all SubgraphInvocations, including the event-backed chat calls from [api][runtime][java][python] Add event-backed chat calls from actions #1220?
• Should the bound be per action execution or per operator subtask?

I'll convert this PR to a draft and keep it for reference until the first split PR is up.

[yunfengzhou-hub]

Hi @yunfengzhou-hub, thanks for the heads-up.

After rebasing, I see that #1138 landed with a different internal-call model, and #1220 then moved internal sub-agent execution onto SubgraphInvocation (removing InternalSubagentCallStatus). So this PR can't be rebased mechanically: the admission and queueing logic needs to be rebuilt on the new invocation model.

Instead of one ~4.9k-line PR, I'd like to split #1140 into smaller PRs:

1. Bounded admission for internal sub-agent invocations in the Java runtime: a concurrency option plus queueing/backpressure on top of SubgraphInvocation, with unit tests.
2. Durable state for queued-but-not-started calls, and recovery of an in-flight batch.
3. Python runtime parity.
4. E2E tests and the Sub-Agents guide.

Two questions before I start:

• Should the concurrency bound apply only to internal sub-agent invocations, or to all SubgraphInvocations, including the event-backed chat calls from [api][runtime][java][python] Add event-backed chat calls from actions #1220?
• Should the bound be per action execution or per operator subtask?

I'll convert this PR to a draft and keep it for reference until the first split PR is up.

Hi @dangzitou, Flink Agents already has configurations that present an upper bound to the concurrency of sub-agent invocations, like num-async-threads or async.batch.parallelism. The "bounded parallelism" requirement of #1140 can be satisfied by these existing limits for now, without introducing a new upper bound — which also means neither of the two questions (scope and granularity) needs to be answered at this point.

What do you think? And if we could set aside the code related to this upper bound, will this PR still be so big that it needs to be divided into multiple ones?

[wenjin272]

My original goal in [#1348](#1348) was to move SubagentTool out of the API and reduce the subagent-specific logic in ToolCallAction. This led me to unify the Future implementations and ctx.gather scheduling, which also enables batching Chat and subagent calls.

This overlaps with the Java batching work here. My preference is to handle execution differences within the Future implementations and share the batch scheduler and limits, without introducing subagent-specific branches in the scheduler.

#1348 is Java-only for now; Python alignment will follow after it merges. Your review would be very welcome, especially around subagent batching and recovery.