[api][plan][runtime] Introduce AGENT resource type and sub-agent invocation API - #938
[api][plan][runtime] Introduce AGENT resource type and sub-agent invocation API#938pltbkd wants to merge 9 commits into
Conversation
9356349 to
875b513
Compare
github-actions
Bot
added
doc-needed
weiqingy
left a comment
weiqingy
left a comment
There was a problem hiding this comment.
Thanks for taking this on. A few questions inline.
| private static final long serialVersionUID = 1L; | ||
|
|
||
| private final boolean success; | ||
| private final Object result; |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
Thanks for catching this — it's a real oversight. I'm currently working on the cross-language sub-agent invocation design and ran into the same issue there.
The plan is to add getResult(Class<T>) backed by OBJECT_MAPPER.convertValue. This unifies all three paths where result can appear as a LinkedHashMap: durable recovery (Jackson JSON deserialization), cross-language sub-agent calls (pemja conversion), and first execution (direct cast, no conversion needed). convertValue handles the Map→POJO conversion uniformly regardless of the source.
One known limitation: for generic collection payloads like List, both recovery and cross-language paths leave result as ArrayList — getResult(List.class) only returns List, losing element types. A getResult(TypeReference<T>) overload can be added later if type-safe collections are needed.
| if call_id is None: | ||
| call_id = ctx.next_call_id(session_id) | ||
| callable_ = self.as_async_callable(ctx, prompt, session_id, call_id) | ||
| return ctx.durable_execute( |
There was a problem hiding this comment.
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).
There was a problem hiding this comment.
We should indeed pass the id to the persistence layer. A simple fix similar to your second suggestion would be to give _invoke an id parameter and pass it as an arg to durable_execute, so that args_digest differentiates calls by their session_id#call_id identity. This works without changing _compute_function_id or durable_execute.
Before settling on a fix though, I'd like to ask for your opinion on a deeper question. Before this PR, Python had no DurableCallable at all. The current "alignment" requires subagent to construct a DurableCallable, but it is a same-named dataclass that is fundamentally different from Java's — it is not actually part of the durable_execute mechanism (_compute_function_id does not recognize it). If we want true cross-language consistency, we should introduce a complete DurableCallable so that _compute_function_id can directly accept it and read its id, mirroring Java's getId().
The other question is whether subagent should expose asDurableCallable at all. In the discussion (#909) I shared my rethinking of the interface design. The current surface was driven by durable_execute binding yield and batch execution, but I'm starting to feel this interface may not be the right one for subagent. I'd appreciate your thoughts on that as well. If we change the interface, the DurableCallable question may not need to be resolved here.
There was a problem hiding this comment.
There is also a DurableCall introduced in #926, maybe we could consider whether to use it and whether to add an id to it (to align with the Java version).
| private final String sessionId; | ||
| private final String callId; | ||
|
|
||
| protected BaseSubagentCallable(String sessionId, String callId) { |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
BaseSubagentCallable doesn't shield any DurableCallable convention — it inherits reconciler() as-is (default null) and only simplifies boilerplate (getId, getResultClass, call with exception capture). Without a reconciler, a crash between invocation and persistence causes replay to re-invoke the external agent — this is expected, identical to any DurableCallable, and typically fine for read-only or idempotent sub-agents. If side effects must not be replayed, the author should provide a reconciler per the existing DurableCallable contract.
The question is whether to surface this option more visibly. Here are three options:
- Add reconcilerInternal to BaseSubagentCallable — simplest, but just restates a convention DurableCallable already defines.
- Remove BaseSubagentCallable — if the abstraction confuses developers into thinking it shields DurableCallable contracts, maybe we should remove it makes them implement DurableCallable directly.
- Default reconciler + abstract querySuccess() — if the typical pattern is if (querySuccess()) return result; else return callInternal(), we can provide a default reconciler() that uses it. Enables the pending/reconcile path; developers can still override.
Which do you prefer?
There was a problem hiding this comment.
Your commits already made the call here. 2 plus 3 is where I'd have landed too, so nothing further from me on this one.
|
|
||
| /** Creates a failed result carrying the full stack trace of the given exception. */ | ||
| public static Result error(Exception exception) { | ||
| return new Result(false, null, exception == null ? null : stackTraceOf(exception)); |
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
This is by design. As discussed in #909, "sub-agent implementations should intercept internal exceptions and populate Result, without directly exposing them to the caller." The intent is to prevent sub-agent exceptions from directly affecting the caller while avoiding the need for every caller to write its own try-catch. When an exception occurs, the persisted Result carries the failure info; the caller always receives a failed Result rather than the exception itself, so Action behavior stays consistent between first run and replay.
On durable-layer status (SUCCEEDED for failed sub-agent calls): This is deliberate. SUCCEEDED at the durable layer means "the durable call completed and returned a Result" — the sub-agent's success or failure is carried inside the Result payload (result.isFailure()), which is a separate concern from durable execution status. On replay, the caller reads the same failed Result from the persisted resultPayload, so processing behavior matches the first run.
On uncapped stack trace: The current text-based exception (full stack trace as errorMessage) was chosen to ensure cross-language transmission. I also felt the full stack trace is more helpful for problem analysis, and since exceptions are typically low-frequency events, storage savings wasn't a primary concern. While I also agree that for the caller the stack trace isn't critical — it's sufficient to find it in logs. I can align with the current convention: persist only exception type + message, and log the full stack trace.
If we were to switch to using DurableExecutionException directly, besides the cross-language issue, there's also a pre-existing problem where the recovered exception (RuntimeException from toException()) differs from the original exception type on first execution — catch blocks targeting specific exception types would no longer match after recovery. This is unrelated to sub-agent; we can discuss it separately.
Does the by-design approach (capturing into Result rather than throwing) seem reasonable, or do you prefer requiring callers to try-catch?
There was a problem hiding this comment.
Capturing into Result rather than throwing seems right to me, and SUCCEEDED at the durable layer reads correctly for the same reason: the record says the call completed and returned a value, and whether that value is a failure is the sub-agent's concern.
The stack-trace half is already handled in 519816d, which logs the full trace and persists just type plus message (Result.java:79,88). Nothing further from me.
| * executions (see the {@code RunnerContext#nextCallId(String)} contract). | ||
| */ | ||
| public String nextCallId(String sessionId) { | ||
| int ordinal = perSessionCallOrdinals.merge(sessionId, 1, Integer::sum); |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
Contract question — direct answer: The RunnerContextImpl warning that "session ids [should] not be shared between action executions" is the correct contract. The current identity context is scoped to a single action task — Subagent.java's "callers may supply a session id to continue a prior session" should be qualified as "within the same action execution," and the RunnerContext interface's nextCallId should also document this scope.
Root cause: when session id / call id counting is reused across action tasks, branch/diamond structures may cause the ordering at first acquisition to differ from the ordering at replay, making sessionId replayability unguaranteeable.
Taking this nit as a starting point, I'd like to discuss what infrastructure external subagent should provide for cross-action session management. Currently nextSessionId / nextCallId are framework built-in tools, and SubagentSetup can already be overridden to use a custom id management mechanism. However, the ordering-induced replayability problem persists in every approach. Here are the directions I see:
Option 1: Doc + tests only
Codify the contract as: "Anonymous Session can by default only continue conversation within the current Action; cross-Action / cross-record conversation continuation requires the Subagent implementation to provide that capability itself." The framework provides no cross-action infrastructure; implementations decide how to manage sessions.
Option 2: Unified override in BaseExternalSubagent
Override nextSessionId / nextCallId in BaseExternalSubagent, maintaining its own sessionId → session-info mapping instead of using the context's built-in tools. This gives external subagents out-of-the-box cross-actionTask session management. The ordering risk remains.
Option 3: getSession interface
Replace the single submit with getSession(sessionId<optional>), then submit on the same session instance to continue the conversation. The session instance internally maintains the ordinal; the subagent manages session instances and eviction timing itself, and the same session can be retrieved from different tasks. This is more natural, but the actual implementation is similar to Option 2 — it introduces a Session object for session-level state, adds complexity for single-call subagents, and still cannot resolve the cross-Action ordering risk.
How far do you think we should go?
There was a problem hiding this comment.
Option 1 sounds right for this PR. The other two add session machinery while admitting they don't fix the ordering risk, which was the reason to build them in the first place.
One thought on where the doc goes. The constraint sits on SubagentIdAllocator.nextCallId (:143) today, but what a caller actually touches is Subagent.submit(ctx, prompt, sessionId) (Subagent.java:37), which just says "Issues an invocation under the given sessionId". BaseSubagentSetup.submit pulls the ordinal from the per-task allocator (:125), and perSessionCallOrdinals starts fresh each task, so passing the same session id from a second action gets you session-1 again.
That matters more in the async mode than it did last round, since the pair is now the remote key (submitRequest.getId() at BaseAsyncSubagentSetup.java:126, and queryStatus / fetchResult probe the same pair). Reusing a session id across two actions would have the second call see the first one's run.
Would putting the "within one action execution" scope on Subagent.submit itself, plus a test, cover it?
|
Hi @weiqingy, Thanks for the review! I've replied to each comment, some actions are being taken. Besides, I've left my reconsideration of the interface in the discussion thread, which may also affect the PR. I would appreciate it if you could take a look and share what you think. |
…ferable contexts record
875b513 to
de21766
Compare
…ministic id assignment
de21766 to
cc114ae
Compare
There was a problem hiding this comment.
Thanks for working through the comments. I re-ran the Python durable-identity check on the new head and the ids now key the durable call properly, and getResult(Class) closes the LinkedHashMap one. Having await fail fast instead of blocking the mailbox where there's no stackful suspension is a nice touch. A few questions on the new surface inline.
|
|
||
| boolean currentInputEventFinished = false; | ||
| if (isFinished) { | ||
| notifyTaskFinished(actionTask); |
There was a problem hiding this comment.
This fires after the action is already recorded as done. maybePersistTaskResult has called actionState.markCompleted() (DurableExecutionManager.java:269), and the output events went downstream at line 437. So checkEmpty throwing here can't undo anything.
On the restart, the action passes actionState.isCompleted() at line 379 and gets skipped, so no handle is ever registered and the check quietly passes. The dropped call ends up skipped either way, just one job failure later.
Would moving the check earlier help, right after actionTask.invoke returns at line 415?
There was a problem hiding this comment.
Makes sense. Besides that, I'll also add the call in the skip branch for already-completed actions, keeping it paired with notifyTaskPrepared, which stays on the common path.
| first and resends only a missing run, so a crash after the POST | ||
| landed never duplicates the prompt. | ||
| """ | ||
| ctx.durable_execute( |
There was a problem hiding this comment.
This one is synchronous. durable_execute is documented as blocking the operator until it finishes (flink_runner_context.py:656), so the whole remote POST sits on the mailbox thread and nothing else moves while it's out.
Java does it async (BaseAsyncSubagentSetup.java:113), and the await path in this same file already uses durable_execute_async at line 142.
Is the sync call intentional here, or should this match the await path?
There was a problem hiding this comment.
Sorry for this mistake. It should use durable_execute_async, I'll handle it.
| } | ||
|
|
||
| /** The durable fetch of a terminal run's result, keyed by {@code sessionId#callId#fetch}. */ | ||
| protected DurableCallable<Result> fetchResult( |
There was a problem hiding this comment.
Nothing in production calls this. awaitResult goes straight to callFetchResult at line 207, so the #fetch slot described at line 60 never gets written. The only caller is fetchResultForTest (MockAsyncSubagentSetup.java:155), and Python is the same shape: fetch_result (async_subagent.py:286) is test-only, and _await_until_terminal calls call_fetch_result at line 322.
That also means the note at 86-89 about persisted fetch records short-circuiting can't happen, since nothing writes one.
I can see why it went this way, awaitResult.call() runs off the mailbox so it can't call back into durableExecuteAsync. Is the plan to make the fetch slot reachable later, or would dropping it for now be cleaner?
There was a problem hiding this comment.
You're right. The fetch slot was reserved for making the fetch a separate durable step, and it lost its writer once the execution moved to the await-driven form. It can be safely dropped and re-added if a concrete need shows up.
| ((DeferredSubagentFuture) future).prepare(); | ||
| } | ||
| } | ||
| // TODO: execute the prepared calls as one batch once durable execution supports batched |
There was a problem hiding this comment.
Three different descriptions of the same behavior. This TODO says serial, the class javadoc at 30-33 says the calls go out together as a batch, and the public SubagentFutures.awaitAll javadoc says each handle resolves when its own wait starts (SubagentFutures.java:44). Callers read that last one.
On the batching itself, #926 adds reservePendingBatch on both RunnerContextImpl and the Python bridge, which looks like the piece this TODO is waiting for. It's still open so nothing can lean on it yet, but have you tried the two together? Curious whether preparing everything up front and handing over one list fits, or whether the deferred mode needs something #926 doesn't expose.
There was a problem hiding this comment.
Sorry about the confusion — I'll polish the code and docs again once we agree on the new surface.
On batching: the plan is indeed to build it on #926, but I haven't verified the fit yet. I'd rather pick it up as a follow-up after #926 lands — batching stays internal to group resolution, so it shouldn't affect the overall design.
| * proceed in between. Its heap state does not survive a failover; the identity is the only basis | ||
| * for rebuilding it through replay. | ||
| * | ||
| * <p>Returned by {@link Subagent#submit}. The invocation is always deferred: the request is issued |
There was a problem hiding this comment.
This says the request always goes out when the handle resolves, but the async mode POSTs during submit (BaseAsyncSubagentSetup.java:113), which its own javadoc calls the pub side.
It shows up in cancel: cancelling a deferred handle sends nothing, cancelling an async one fires a remote cancel. Someone holding a SubagentFuture can't tell which they have from this text.
Could this be scoped to the deferred mode?
There was a problem hiding this comment.
Yes, that paragraph describes the deferred mode. It predates the async base and wasn't updated when that landed, I'll align the doc to the final state.
| RunStatus probe = setup.queryStatus(getSessionId(), getCallId()); | ||
| return probe.getState() == RunStatus.State.COMPLETED | ||
| || probe.getState() == RunStatus.State.FAILED; | ||
| } catch (Exception e) { |
There was a problem hiding this comment.
A failing probe turns into false with nothing logged. If queryStatus keeps failing, say expired credentials or a dead endpoint, isDone() reports "not done" forever and there's no clue why. Result.error does log on the same kind of path (Result.java:79). Worth a debug log here?
There was a problem hiding this comment.
Agreed, the swallow is not right here.
The contract should be: the implementation handles the failures queryStatus can understand and reports them as a FAILED status, your examples fall in this category; an exception that still escapes is a system-level failure the implementation can't handle.
While we generally want sub-agent exceptions kept away from the main agent, swallowing such an exception (even with a log) leaves the await waiting forever, while converting it into a failed result may diverge from what the remote side actually did. For this version, the safest choice is to throw. I plan to drop throws along the queryStatus path and the try-catch here, and ask implementors to report such failures as a RuntimeException.
There was a problem hiding this comment.
Agreed on failing rather than hanging.
One case I'm not sure your plan reaches: awaitResult catches Exception at BaseAsyncSubagentSetup.java:217-219, so dropping throws would leave that one still turning a probe failure into a Result. Is await() meant to stay lenient there, or would you want it following isDone()?
(reconcileSubmitRequest already propagates, so that one looks fine.)
Python's done() swallows the same way (async_subagent.py:123-126). Would that move with it?
| public abstract class BaseAsyncSubagentSetup extends BaseSubagentSetup { | ||
|
|
||
| /** Delay between status probes while waiting for the run to reach a terminal state. */ | ||
| protected long statusPollIntervalMillis = 10; |
There was a problem hiding this comment.
nit: 10ms works out to roughly 100 status calls a second per run, against the services named at lines 34-35 (LangGraph, OpenAI Assistants, A2A), which don't usually finish that fast. Tests set it to 0 (MockAsyncSubagentSetup.java:71), so integrations inherit this value. Would a bigger default, or a backoff, fit better?
There was a problem hiding this comment.
I'll make it 500 by default. This value affects not only the probe frequency but also the latency bound. Since it's sensitive on both ends, would it be worth introducing an option to encourage users to overwrite it for their own service?
There was a problem hiding this comment.
Implementors can already change it. statusPollIntervalMillis is a plain protected field (BaseAsyncSubagentSetup.java:100). So the ones who'd benefit from an option are users picking up someone else's setup class, since nothing here reads the resource descriptor. Is that the group you had in mind?
If so, would a descriptor argument fit better than a config option? AnthropicChatModelConnection.java:98-106 takes timeout / max_retries that way and Ollama/Gemini/Bedrock follow it, and Python would pick it up for free (resource_provider.py:118).
nit: should Python go to 0.5 alongside the 500? async_subagent.py:234 calls itself the parity of statusPollIntervalMillis.
| * the request here (an internal sub-agent sends its call event); the returned callable's {@link | ||
| * DurableCallable#call()} carries only the part that runs off the mailbox thread. | ||
| * | ||
| * <p>Implementations that recover an in-flight invocation after failover supply the reconciler |
There was a problem hiding this comment.
nit: following up on the reconciler question from last round. The async base now wires one by default, and leaving it to the implementor here seems right, since the deferred mode isn't tied to external services.
Would it be worth spelling out what skipping it costs? Something like: without a reconciler, a crash between the call landing and its result being persisted re-invokes on replay. Then it reads as a choice rather than a default.
| /** | ||
| * Issues an invocation under the given {@code sessionId}; the implementation picks the call id. | ||
| */ | ||
| SubagentFuture submit(RunnerContext ctx, Object prompt, String sessionId) throws Exception; |
There was a problem hiding this comment.
nit: the PR description still describes the old surface. asAsyncCallable, callAsync and executeAllAsync have no occurrences left in the branch, and call() is gone from this interface. Worth refreshing it so the body matches cc114aef?
| * implementation-side contract, declared by {@link SubagentSetup}; resolving a returned handle is | ||
| * {@code await}. | ||
| */ | ||
| public interface Subagent { |
There was a problem hiding this comment.
In the existing framework, an Agent is the thing a user authors. A user builds it by adding actions and resources (chat models, tools, and so on), and it runs in the event driven model. So today "agent" means an authored, event driven unit.
The new Subagent interface is different in nature. It declares only submit, which is a caller side capability: it is how you invoke a remote agent and get a Result back, not something a user authors.
Putting these together, we now have two public types with "agent" in the name that mean quite different things: Agent (authored, event driven) and Subagent (invoked, request response), and the two have no relationship in the type graph. I worry this is confusing users to read, since it is hard to tell whether Subagent is a kind of agent or the handle used to call one.
So my question is whether we should keep a public Subagent interface at all, or express submit on a caller or handle abstraction whose name reflects that it is a way to invoke rather than a kind of agent.
There was a problem hiding this comment.
I view Subagent as a resource rather than a kind of agent: like MCPServer and Tool, the class name is the entity name. I split it off from SubagentSetup mainly because it's the user-facing surface — callers only see submit(), while the full-identity form and the resource plumbing stay on SubagentSetup.
If the split itself is what reads as confusing, I'm fine merging it back into SubagentSetup and having callers use that directly — that's what BaseChatModelSetup does today.
I'd also lean against a handle abstraction here: resources are used through direct method calls on the instance from the context, so a handle would only add a layer — the same reason we don't have ChatModelChatter.chatTo(model) or ToolUser.use(tool).
There was a problem hiding this comment.
Yeah, as we discussed in the meeting, drop Subagent and keep SubagentSetup as front interface is more user friendly - reduce user confusion between Agent and SubAgent.
| * message — rather than a live exception, so that a {@code Result} can be persisted through durable | ||
| * execution. The full stack trace is logged when the failure is captured, not persisted. | ||
| */ | ||
| public class Result implements Serializable { |
There was a problem hiding this comment.
Looks like all other API names have Subagent as prefix, maybe we should follow similar pattern for this class - SubagentResult?
| * Caller-facing interface for all sub-agents (external and internal). | ||
| * | ||
| * <p>An invocation is identified by a {@code (sessionId, callId)} pair; the session groups a | ||
| * conversation across invocations. Callers do not manage ids: the short forms below leave the |
There was a problem hiding this comment.
Callers do not manage ids: the short forms below leave the missing ids to the implementation,
Looks like this is not consistent with the method definition below, it still requires the session id passed in:
SubagentFuture submit(RunnerContext ctx, Object prompt, String sessionId) throws Exception;
There was a problem hiding this comment.
"Do not manage" here doesn't mean "never supply": the form taking a sessionId is kept to continue a prior conversation, and that id still doesn't need caller management — it can be obtained from the future returned by an earlier id-less call (SubagentFuture.getSessionId()). For simple invocations the id-less submit is enough. I'll reword the javadoc to spell this out.
There was a problem hiding this comment.
Yeah, let's rephrase it a little bit to make it more clear. We can do this in SubagentSetup class since we drop this class.
3 participants
Design discussion: #909
Purpose of change
Introduces
AGENTas a first-class resource type and the caller-facing sub-agent invocation API across Java / Python / YAML, per discussion #909.SubagentSetupis the general sub-agent framework. This PR lands the framework and the one way to plug into it today: a customSubagentSetupimplementation. Registering anAgentdirectly as an internal sub-agent is a follow-up addition.Included:
ResourceType.AGENT; register viaaddResource(name, AGENT, setup)and via YAMLsubagents:.Subagent(caller interface) +SubagentSetup(base) +Result+BaseSubagentCallable(captures failures intoResult).call()/asAsyncCallable()run through durable execution; a deterministic(sessionId, callId)identity lets failover replay reuse cached results instead of re-invoking. UseasAsyncCallable()instead ofcallAsync()to support batch async execution.sessionIdis framework-generated and deterministic; callers may pass their own to continue a session.callIdis auto-generated from the per-session conversation ordinal.Evolved from the #909 proposal during review:
Resultcarries a serializableerrorMessage(the failure's full stack trace) instead of a liveException, so it survives durable persistence.(sessionId, callId).Notes
Agentdirectly as anAGENTresource, compiled into a scoped child plan, and its isolated execution).Resultand the parallel async primitive (executeAllAsync) overlap with Parallel Tool Execution ([Feature] Parallel Tool Call Execution #926) and can be aligned/iterated during review.Tests
SubagentSetup+ YAML descriptor), and deterministic-id context / failover-recovery / operator-integration tests.ExternalSubagentTest— programmatic and YAML-declared sub-agent invocation (including failure surfaced viaResult) 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 YAMLsubagents:block.No breaking changes to existing APIs.
Documentation
doc-neededbut deferred: add once the internal sub-agent lands and the API stabilizesdoc-not-neededdoc-included