[api][plan][runtime] Introduce AGENT resource type and sub-agent invocation API

[pltbkd]

Design discussion: #909

Purpose of change

Introduces AGENT as a first-class resource type and the caller-facing sub-agent invocation API across Java / Python / YAML, per discussion #909.

SubagentSetup is the general sub-agent framework. This PR lands the framework and the one way to plug into it today: a custom SubagentSetup implementation. Registering an Agent directly as an internal sub-agent is a follow-up addition.

Included:

• ResourceType.AGENT; register via addResource(name, AGENT, setup) and via YAML subagents:.
• Subagent (caller interface) + SubagentSetup (base) + Result + BaseSubagentCallable (captures failures into Result).
• call() / asAsyncCallable() run through durable execution; a deterministic (sessionId, callId) identity lets failover replay reuse cached results instead of re-invoking. Use asAsyncCallable() instead of callAsync() to support batch async execution.
• sessionId is framework-generated and deterministic; callers may pass their own to continue a session. callId is auto-generated from the per-session conversation ordinal.

Evolved from the #909 proposal during review:

• Result carries a serializable errorMessage (the failure's full stack trace) instead of a live Exception, so it survives durable persistence.
• The durable id is derived solely from (sessionId, callId).

Notes

• Follow-up additions: ChatModel calling sub-agents (sub-agent-as-tool dispatch), and internal sub-agent (registering an Agent directly as an AGENT resource, compiled into a scoped child plan, and its isolated execution).
• Cross-language execution (an action calling a sub-agent implemented in the other language) is not supported yet; registration round-trips across the plan-JSON boundary, execution will be added later.
• Timeout / cancellation semantics are not included yet; planned as a follow-up.
• Result and the parallel async primitive (executeAllAsync) overlap with Parallel Tool Execution ([Feature] Parallel Tool Call Execution #926) and can be aligned/iterated during review.

Tests

• Java unit: AGENT registration, plan compilation (SubagentSetup + YAML descriptor), and deterministic-id context / failover-recovery / operator-integration tests.
• Python unit: mirrors the Java surface (registration, plan resources, id delegation).
• e2e: ExternalSubagentTest — programmatic and YAML-declared sub-agent invocation (including failure surfaced via Result) on a real Flink job.

API

Additive public API, kept semantically aligned across Java / Python / YAML: ResourceType.AGENT, Subagent, SubagentSetup, Result, BaseSubagentCallable, RunnerContext#nextSessionId() / nextCallId(sessionId), and the YAML subagents: block.

No breaking changes to existing APIs.

Documentation

• doc-needed but deferred: add once the internal sub-agent lands and the API stabilizes
• doc-not-needed
• doc-included

[weiqingy]

Thanks for taking this on. A few questions inline.

[weiqingy]

result is typed Object, BaseSubagentCallable.getResultClass() pins the durable result class to Result.class (BaseSubagentCallable.java:47-48), and recovery re-binds through the plain OBJECT_MAPPER at RunnerContextImpl.java:516, which is constructed with no polymorphic typing (:67-68).

I round-tripped a small POJO payload through this Result with writeValueAsString / readValue(s, Result.class):

serialized: {"success":true,"result":{"verdict":"approve","score":7},"errorMessage":null}
payload class after replay: java.util.LinkedHashMap
ClassCastException: class java.util.LinkedHashMap cannot be cast to class Review

So getResult() hands back the author's type on the first execution and a LinkedHashMap after a failover replay. The shipped example casts at ExternalSubagentAgent.java:52 (((List<?>) result.getResult()).get(0)) and survives only because JSON arrays bind to ArrayList. Every payload in the suite is a String or a List<String> (MockExternalSubagentSetup.java:92, SubagentIdentityRecoveryTest.java:108), so nothing currently exercises the shape that breaks.

Python does not diverge here. Its durable payload goes through cloudpickle (flink_runner_context.py:430,473), which preserves the type, so this is also a Java/Python semantic gap on new public API that AGENTS.md asks to keep aligned.

What should getResult() return after a replay when the sub-agent returned a record or a POJO? A couple of routes, in case they help: making Result generic and threading the payload class through getResultClass(), or keeping the field opaque and adding getResult(Class<T>) backed by OBJECT_MAPPER.convertValue. Either way a test with a non-String, non-collection payload would pin the behavior.

[weiqingy]

durable_execute receives callable_.call, so the DurableCallable and its id are dropped at this boundary. durable_execute has no id parameter in either the ABC (runner_context.py:200-207) or the implementation (flink_runner_context.py:639-646), and keys the durable call on _compute_function_id(func) plus _compute_args_digest(args, kwargs) (flink_runner_context.py:415-416).

_invoke is defined on BaseSubagentCallable (:99) and bound as call in __init__ (:97), so __qualname__ is identical for every subclass. I loaded this file's class hierarchy and ran the real _compute_function_id / _compute_args_digest over three callables with different sub-agents, sessions and call ids:

ReviewerCall  id=sA#c1  function_id=flink_agents.api.subagent.BaseSubagentCallable._invoke  args_digest=f744dd20f806e514
ReviewerCall  id=sB#c9  function_id=flink_agents.api.subagent.BaseSubagentCallable._invoke  args_digest=f744dd20f806e514
CoderCall     id=sZ#c3  function_id=flink_agents.api.subagent.BaseSubagentCallable._invoke  args_digest=f744dd20f806e514

args is always empty on this call path, so the digest is constant as well. Every sub-agent durable call in a Python job collapses to one (function_id, args_digest) pair, and only the positional currentCallIndex (RunnerContextImpl.java:789-802) tells two calls apart.

Two consequences, plus one weaker one. The PR's headline property, that a deterministic (sessionId, callId) lets failover replay match the right cached result, holds on Java only; Python still matches by call ordinal exactly as before. And DurableCallable.id is public surface in Python whose docstring at :73-77 describes a key nothing reads, its only reader anywhere being the assertion at test_subagent.py:145. The weaker one: Java's matchNextOrClearSubsequentCallResult mismatch guard cannot fire for Python sub-agent calls, though divergent replay is already documented as undefined behavior at runner_context.py:218-220.

Was the id meant to reach durable_execute here? Adding an id / call_id parameter to durable_execute and durable_execute_async and forwarding it as the function_id is one way; passing a per-call uniquely-named wrapper instead of the bound method is another. A Python replay test would catch the regression either way, and the harness already exists at python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py (fake Java context with matchNextOrClearSubsequentCallResult and recordCallCompletion).

[weiqingy]

Reading this from the perspective of someone writing an external integration against the new surface. BaseSubagentCallable is the convenience base the API steers implementations to (:23-29, and subagent.py:199-201 says so explicitly). It does not override DurableCallable#reconciler(), so every sub-agent callable inherits the null default at DurableCallable.java:74.

durableExecute selects the reconcile state machine only when reconciler() is non-null (RunnerContextImpl.java:267-275), and durableExecuteAsync, which is the path SubagentSetup.call takes, gates identically (JavaRunnerContextImpl.java:62-70). Either way sub-agent calls land on durableExecuteCompletionOnly. On that path appendPendingCall is never reached: its only callers are inside durableExecuteWithReconcile (:555, :561). A crash between "external agent invoked" and "result persisted" therefore leaves no record at all, replay misses the cache, and call() re-invokes the external agent.

Grepping the new surface, reconcil does not appear anywhere under api/.../subagent/, in the e2e tests, or in the runtime sub-agent tests. Python surfaces the field (subagent.py:82) and forwards it (:184), but wires None and no test asserts the forwarding. The recovery tests seed only terminal CallResults (SubagentIdentityRecoveryTest.java:105-108,171-178), so the pending path has no sub-agent coverage on either side.

Concretely: what does an integration author write today to get reconcile-before-resend, and is there anything on this surface that would tell them the option exists? One shape that would make the choice visible, in case it is useful: BaseSubagentCallable taking the reconciler as a constructor argument, so passing null is something the author decided rather than a default they never saw.

[weiqingy]

error(Exception) stores the full stack trace as errorMessage, and BaseSubagentCallable.call() captures every exception into it rather than throwing (BaseSubagentCallable.java:52-58). Two things follow downstream.

The durable layer sees a normal completion: durableExecuteCompletionOnly calls recordDurableCompletion with a null exception (RunnerContextImpl.java:329), and CallResult's status is derived as exceptionPayload == null ? SUCCEEDED : FAILED (CallResult.java:100). So isFailure() (:177-179) is false for every failed sub-agent call, and getCurrentCallResultFields() reports "SUCCEEDED" (RunnerContextImpl.java:490). Anything keyed on durable-call status reads sub-agent failures as successes.

The trace is also uncapped, and recordCallCompletion persists the ActionState immediately to the configured store (RunnerContextImpl.java:827-837, backed by KafkaActionStateStore / FlussActionStateStore). A flapping external agent writes a multi-KB string per failure into durable storage.

Worth capping what gets persisted, say the message plus the top N frames, and keeping the full trace to the log? And is collapsing a captured failure into a SUCCEEDED CallResult deliberate, or should the two stay distinguishable at that layer?

[weiqingy]

nit: the javadocs disagree about whether a caller-supplied session id is safe, and both examples take the side this one warns against.

perSessionCallOrdinals is per-task heap state, so this ordinal restarts at 1 for any session id the task has not seen. The javadoc directly above says cross-task uniqueness "relies on session ids not being shared between action executions (see the RunnerContext#nextCallId(String) contract)". The contract it points at states no such thing: RunnerContext.java:152-153 reads in full, "Creates a new call id for a sub-agent invocation under the given session." Subagent.java:27-30 goes the other way again, saying callers may supply a session id to continue a prior session.

The suite shows the effect without asserting on it. it1MixedCallsProduceUniqueDeterministicIds (key 1L) and it1RerunningIdenticalInputReproducesIdenticalCaptureSequence (key 2L) both produce explicit-session-checkout-1-1 (SubagentIdentityIntegrationTest.java:102), because ExternalSubagentAgent.java:50 passes "session-" + prompt and external_subagent_agent.py:82 is identical.

Nothing breaks today. CallResults live inside an ActionState already scoped by ActionStateUtil.generateKey (:44-55), so two executions never share a list, and neither example passes the id to its endpoint. It would start to matter if an integration used sessionId#callId as the remote-side identity, which seems a natural reading of a durable call id.

Which javadoc is the contract? If the uniqueness obligation sits on the caller, Subagent's javadoc could say so, and the examples could stop modelling the pattern this one warns against.