[wenjin272]

Linked issue: Closes #1241

Purpose of change

Sub-agents can be declared through ordinary resource descriptors while preserving descriptions and input schemas for model-selected calls. This removes the dedicated internal-subagent provider and the different Java/Python serialized-resource representations.

Runtime flow

1. InternalSubagentResourceDescriptorBuilder().build() returns a descriptor (constructed with new in Java). Agent.addResource / add_resource also wraps the direct-Agent shorthand in a descriptor.
2. Plan compilation replaces the Agent argument with compiled JSON data under plan, assigns the registration name as scope, and retains recursive cycle detection.
3. Generic providers construct runtime setups. Java owns internal workflow scheduling; a Python caller uses the runtime adapter to create its local invocation handle.
4. When internal setups and a Python bridge exist, the operator forwards task lifecycle events so lazy Python handles can allocate call identities. Bootstrap, awaiting, and result collection retain their existing execution paths.

Key decisions

Keep the generic ResourceDescriptor unchanged. The builder supplies declaration convenience; the plan layer owns compilation, and runtime owns invocation. SubagentSetup extends Resource, since the plan transfers its descriptor rather than a live setup instance.

Behavioral Semantics

Interaction decisions

Declaration / caller	Result
Internal descriptor or Agent shorthand / Java	Generic Java provider constructs the internal setup.
Internal descriptor or Agent shorthand / Python	Same Java-owned plan; Python obtains a local runtime handle and lifecycle notifications.
External descriptor / same-language caller	Generic provider constructs the declared setup.
External Python descriptor / Java	Existing Python-owned runtime bridge path is retained.
External Java descriptor / Python	Unsupported; raises NotImplementedError.
Internal builder with input schema / model-selected call	Description and schema reach the sub-agent callable; shorthand alone supplies no schema.

Behavioral contracts

• Direct-Agent registration remains available; the child is compiled when its parent plan is built.
• Description/schema survive compilation and JSON transfer without modifying the original declaration. Input types produce schema data, not automatic payload conversion.
• A child registered under multiple names retains a distinct scope per name. Recursive declarations are rejected.
• Internal calls still support output collection, failure results, and nested calls; model tool requests can execute an internal descriptor after plan transfer.
• External setup configuration is reconstructed from a descriptor; live setup instances are unsupported.

Failure behavior

Builders reject missing Agents, conflicting schema/type declarations, unsupported input types, and blank schemas. Plan compilation rejects cycles and invalid AGENT declarations; runtime decoding rejects missing or malformed compiled plan data. These errors raise without fallback or retry. Child action failures retain the existing failed-result path. External transport, retry, and recovery semantics are unchanged.

Tests

Contract	Coverage
Direct-Agent shorthand and normal descriptor construction	AgentPlanSubagentResourceTest, test_agent_plan_subagent_resources.py
Metadata/schema and declaration immutability	InternalSubagentResourceDescriptorBuilderTest, descriptorMetadataSurvivesCompilationAndJsonWithoutMutatingDeclaration, test_internal_subagent_descriptor.py
Distinct alias scopes and cycle rejection	Java/Python subagent resource tests, descriptor cycle tests
Plan JSON in both language directions	Java/Python AgentPlanCrossLanguageTest suites and internal-subagent snapshots
Output, failure, nesting, and model tool execution	InternalSubagentCallTest, InternalSubagentDescriptorTest, three real Flink internal E2E cases
External descriptor reconstruction	Java serialization/resource tests, Python round-trip test, three real Flink external E2E cases
Setup-instance rejection	SubagentRegisterTest, Python plan compilation rejection test

Passed: Java API/Plan/Runtime reactor (2,494 passed, 44 skipped), Python API/Plan/Runtime (1,723 passed, 15 skipped), and six real Flink E2E tests using freshly rebuilt JARs. Ruff, Java 11 Spotless, source-export Apache RAT, and whitespace checks passed.

Python 3.12 regression verification: reproduced the runtime descriptor test failure with an unconstrained mock, then verified the protocol-constrained fixture with 25 descriptor, registration, and sub-agent API tests on both Python 3.11 and 3.12.

After the builder rename: Java focused tests (43 passed, 1 JDK guard test skipped on JDK 21) and Python 3.12 focused tests (49 passed, 1 snapshot-regeneration test skipped) passed.

Not verified: live remote agent services, old serialized-plan/checkpoint migration, and the full Flink/JDK version matrix. Runtime and E2E execution used JDK 21; existing low-JDK restrictions remain.

Implementation invariants and verification commands

Compiled internal descriptors use JavaResourceProvider in both languages. They contain JSON-compatible plan data and no live Agent. Python decoding copies that data before the existing AgentPlan validator rebuilds providers. Bridge discovery traverses nested descriptor plans; lifecycle registration no longer depends on the removed serialized-provider representation.

• Java: mvn test -pl runtime -am -Dspotless.skip=true (JDK 21).
• Python: pytest flink_agents/api flink_agents/plan flink_agents/runtime -m 'not integration' with the active environment's site-packages on PYTHONPATH.
• Rebuilt dist/common and dist/flink-2.3; ran internal_subagent_test.py, subagent_external_integration_test.py, and subagent_integration_test.py on Flink 2.3.
• License verification used the current source export, excluding ignored local configuration.

API

Adds Java/Python InternalSubagentResourceDescriptorBuilder, constructed directly and returning an ordinary ResourceDescriptor from build(). SubagentSetup now extends Resource; callers registering setup instances must register descriptors instead. Direct-Agent shorthand remains. Internal descriptor data uses plan, with Java setup accessors getPlan() / getPlanJson() and Python plan. Old dedicated-provider JSON is not supported by the new representation. Resource type and generic descriptor APIs are 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.153.4 (GPT-6)

[wenjin272]

Could you please review this PR when you have a chance? @pltbkd @eemario

[pltbkd]

Thanks for the PR. LTGM.

nit: the InternalSubagent class may mislead that it's a 'subagent' rather than a 'ResourceDescriptor' buidler, WDYT?