[Feature] Sub-agent Resource for Flink Agents - Framework part

[pltbkd]

[Feature] Sub-agent Resource for Flink Agents

Motivation

As discussed in #660, there are scenarios where a Flink Agent needs to call external agents — for example, delegating a subtask to a remote LLM agent service or a third-party agent platform. Currently, users can only do this by manually writing remote calls inside actions, which means they miss out on durable execution, checkpoint recovery, and tool call integration.

Beyond external agent calls, several related scenarios have emerged:

• Parallel exploration: Multiple agents working in different directions simultaneously, each with its own reasoning loop and tools.
• Independent review: An agent providing an independent perspective — e.g., a reviewer agent that evaluates code without being influenced by the main agent's context.
• Reusing agents as sub-links: Composing an agent defined by another team as a building block in a larger agent pipeline.

These scenarios also create demand for internal sub-agents — the ability to register a Flink Agents Agent as a sub-agent of another Agent, with proper context isolation and execution semantics.

This proposal introduces AGENT as a first-class resource type in Flink Agents, supporting both internal and external sub-agents with a unified API.

Goals

• Introduce AGENT resource type: Sub-agents can be registered via addResource(name, AGENT, ...) and accessed via ctx.getResource(name, AGENT), usable by both Actions and ChatModel.
• Support for external, internal, and Remote Agents: Supports custom external agents, supports qualified internal agents, and will later implement remote parallel execution of internal agents based on RpcOperator.
• Context isolation, DurableExecute, and failover recovery: Both internal and external sub-agents benefit from durable execution with automatic failover recovery. Internal agents introduce scope-level isolation.

The proposed changes have been initially validated with a POC (mainly on java api). The overall design still has room for discussion — feedback and suggestions are welcome.

---

Proposed Interface

1. Resource Type

public enum ResourceType {
    CHAT_MODEL, CHAT_MODEL_CONNECTION, TOOL, EMBEDDING_MODEL,
    VECTOR_STORE, PROMPT, MCP_SERVER, SKILLS,
    AGENT  // Sub-agent resource
}

2. Subagent Interface

Subagent is the common interface for all sub-agents. Both external (SubagentSetup) and internal sub-agents implement it.

public class Result {
    boolean success;        // success or failure
    Object result;       // type depends on agent implementation; built-in agents return List<Object>
    Exception exception; // present when success=false
}

public interface Subagent {
    default Result call(RunnerContext ctx, Object prompt) {
        return call(ctx, prompt, ctx.nextSessionId());
    }

    default Result call(RunnerContext ctx, Object prompt, String sessionId) {
        return ctx.durableExecuteAsync(asAsyncCallable(ctx, prompt, sessionId));
    }

    default DurableCallable<Result> asAsyncCallable(RunnerContext ctx, Object prompt) {
        return asAsyncCallable(ctx, prompt, ctx.nextSessionId());
    }

    DurableCallable<Result> asAsyncCallable(RunnerContext ctx, Object prompt, String sessionId);
}

class Result:
    success: bool
    result: Any 
    exception: Optional[Exception]

class Subagent(Protocol):
    def call(self, ctx: RunnerContext, prompt: Any, session_id: Optional[str] = None) -> Result:
        if session_id is None:
            session_id = ctx.next_session_id()
        return ctx.durable_execute_async(
            self.as_async_callable(ctx, prompt, session_id))

    def as_async_callable(self, ctx: RunnerContext, prompt: Any, session_id: Optional[str] = None) -> DurableCallable[Result]:
        if session_id is None:
            session_id = ctx.next_session_id()
        ...

SubagentSetup is the abstract base for external sub-agents, implementing Subagent:

public abstract class SubagentSetup extends SerializableResource implements Subagent {

    @Override
    @JsonIgnore
    public ResourceType getResourceType() { return ResourceType.AGENT; }
}

class SubagentSetup(SerializableResource, Subagent):
    def get_resource_type(self) -> ResourceType:
        return ResourceType.AGENT

3. Registration

Internal and external sub-agents are registered identically, only via addResource, not via annotation:

// Internal sub-agent
addResource("reviewer", ResourceType.AGENT, new ReviewerAgent());

// External sub-agent
addResource("summarizer", ResourceType.AGENT, new HttpSubagentSetup(endpoint));

# Internal
self.add_resource("reviewer", ResourceType.AGENT, ReviewerAgent())

# External
self.add_resource("summarizer", ResourceType.AGENT, HttpSubagentSetup(endpoint))

agents:
  - name: reviewer-agent
    module: com.example
    clazz: ReviewerAgent

  - name: root-agent
    module: com.example
    clazz: RootAgent
    subagents:
      - name: reviewer
        agent_ref: reviewer-agent          # internal: reference to another agent definition
      - name: summarizer
        target_module: com.example          # external: reference to SubagentSetup subclass
        target_clazz: HttpSubagentSetup
        arguments:
          endpoint: "http://summarizer:8080"

4. Calling

// Synchronous
Subagent subagent = ctx.getResource("reviewer", ResourceType.AGENT);
Result result = subagent.call(ctx, prompt);

// Parallel
DurableCallable<Result> c1 = subagent1.asAsyncCallable(ctx, prompt1);
DurableCallable<Result> c2 = subagent2.asAsyncCallable(ctx, prompt2);
List<Result> results = ctx.durableExecuteAllAsync(c1, c2);

# Synchronous
subagent: Subagent = ctx.get_resource("reviewer", ResourceType.AGENT)
result: Result = subagent.call(ctx, prompt)

# Parallel
c1 = subagent1.as_async_callable(ctx, prompt1)
c2 = subagent2.as_async_callable(ctx, prompt2)
results: list[Result] = await ctx.durable_execute_all_async(c1, c2)

5. Implementing External Sub-agent

public class HttpSubagentSetup extends SubagentSetup {
    private final String endpointUrl;

    public HttpSubagentSetup(String endpointUrl) {
        this.endpointUrl = endpointUrl;
    }

    @Override
    public DurableCallable<Result> asAsyncCallable(RunnerContext ctx, Object prompt, String sessionId) {
        return () -> {
            List<Object> output = httpClient.post(endpointUrl,
                    Map.of("prompt", prompt, "session", sessionId));
            return new Result(true, output, null);
        };
    }
}

class HttpSubagentSetup(SubagentSetup):
    def __init__(self, endpoint: str):
        self._endpoint = endpoint

    def as_async_callable(self, ctx: RunnerContext, prompt: Any, session_id: str) -> DurableCallable[Result]:
        def _call() -> Result:
            output = httpx.post(self._endpoint, json={"prompt": prompt, "session": session_id})
            return Result(success=True, result=output, exception=None)
        return DurableCallable(_call)

---

Execution Process

1. Compilation

Both internal and external sub-agents compile to SerializableResourceProvider in the plan JSON, ensuring cross-language interoperability:

AGENT resource processing
  │
  ├── SubagentSetup (external)
  │     → SerializableResourceProvider
  │
  └── Agent (internal)
        → Compiled into child AgentPlan
          → InternalSubagentSetup → SerializableResourceProvider

Both paths produce SerializableResourceProvider, resulting in a uniform plan JSON representation with no runtime difference.

Additionally, the compiler registers built-in sub-agents: if CHAT_MODEL is available, ReActAgent is automatically registered as "general-purpose". This registration runs after all user-declared resources are extracted, ensuring decorator-declared CHAT_MODEL is ready.

2. Execution Model

The default call() implementation executes the DurableCallable returned by asAsyncCallable via durableExecuteAsync, leveraging the durable execution mechanism for state recording and recovery.

Parallel calls: users manually split — call asAsyncCallable for each sub-agent, then pass all DurableCallables to durableExecuteAllAsync (to be introduced in an upcoming discussion).

call() returns Result; sub-agent implementations should intercept internal exceptions and populate Result, without directly exposing them to the caller. tool_call_action constructs success/error ToolResponseEvent based on Result, without try/catch.

3. ChatModel Integration

Sub-agents are exposed to the LLM as tools (Subagent as Tool pattern).

• Prompt injection: Available AGENT resources are listed in the system prompt alongside tool descriptions, letting the LLM know which sub-agents it can delegate to.
• Dispatch: tool_call_action checks resource type — TOOL resources go to tool.call(), AGENT resources go to subagent.call(ctx, prompt).
• Result: Returned via existing ToolResponseEvent.

LLM response → tool_calls → ToolRequestEvent → tool_call_action
  │
  ├── name matches TOOL resource → tool.call(**kwargs)
  └── name matches AGENT resource → subagent.call(ctx, prompt)
  │
  └── ToolResponseEvent → back to LLM

LLM APIs (OpenAI, Ollama, etc.) only have tool/function call concepts — there is no "agent call". Sub-agents are exposed as tools with a single prompt parameter. The tool_call_action already holds ctx as an action parameter, which is passed directly to subagent.call(ctx, prompt).

---

Internal Sub-agent Design Preview

Internal sub-agent compilation — registering Agent instances as sub-agents — will be discussed in a future proposal. The resource model and calling API are already designed to accommodate it.

1. Compilation

The sub-agent's Agent is compiled into an independent AgentPlan (childPlan), wrapped as InternalSubagentSetup(childPlan, scope). The child plan inherits the parent plan's AgentConfiguration. Circular references are detected at compile time. scope is the sub-agent's resource name, used for event routing.

2. Invocation

Calling a sub-agent emits a SubagentCallEvent. During async execution, the coroutine yields, and the AEO processes the event in a subsequent mailbox cycle, scheduling the sub-agent's internal actions. Only the event sending timing is modified for SubagentCallEvent to support this execution model — asAsyncCallable sends SubagentCallEvent on the mailbox thread, and durableExecuteAsync blocks on the worker thread waiting for the result.

3. Scope-based Execution

SubagentCallEvent is converted to InputEvent within the sub-agent's scheduling scope. Intermediate events are dispatched only within that scope — they do not leak to the parent agent or other sub-agents. The result OutputEvent is intercepted and returned to the caller. When all pending events are processed and no actions are running (quiesce), the call completes automatically.

4. Isolation

SubagentRunnerContext provides isolated context:

• Actions and events: Sub-agent uses its own AgentPlan's definitions, fully independent from parent.
• Memory: Read-only access to parent memory; Sensory memory is allowed but not written back to parent scope memory.
• Resources: Uses own registered resources first, falls back to parent (parent resource cache inheritance).
• Config: Inherits parent plan's AgentConfiguration.

Nested sub-agents are supported recursively via parent resource cache fallback.

5. Fault Tolerance

• Completed calls: Intercepted directly by DurableExecute mechanism; failover hits cached result.
• Incomplete calls: Supports ActionState; after failover, deterministic SessionId associates the new sub-agent call with recovered Actions, enabling intermediate state recovery rather than full re-execution.

---

Design Discussion

1. Internal vs External Capability Parity

External and internal sub-agents provide identical user-facing API (registration, call(), asAsyncCallable()). However, the underlying implementations differ significantly — external sub-agents are black-box calls (HTTP/gRPC), while internal sub-agents are framework-internal plan nesting execution.

Capability	External Sub-agent	Internal Sub-agent
Context isolation	Managed by external system via deterministic SessionId	Framework creates isolated SubagentRunnerContext
Memory	Maintained by external system	Read-only parent access, write-isolated, not persisted
State recovery	DurableExecute persists call records	Same as external; additionally, deterministic SessionId links call chain to sub-agent internal action ActionState, supporting intermediate state recovery
Parallel execution	Multiple agents in parallel via durableExecuteAllAsync	One action at a time per ActionExecutionOperator, but coroutine-based async partial parallelism
Backpressure	Via durableExecuteAsync worker pool	Same as external; needs to consider subagent nested call deadlock issues

2. Introducing SubagentCompatible Marker

When an Agent is registered as a sub-agent, it runs in an isolated SubagentRunnerContext, meaning certain capabilities are restricted, such as the inability to write to memory. Not all Agents can correctly accommodate these constraints.

To ensure registration correctness, we propose a SubagentCompatible marker (interface / annotation / yaml field) that an Agent declares to indicate it is safe to run as a sub-agent. The framework validates this marker at compile time: when an unmarked Agent is registered as a sub-agent, compilation fails fast.

This provides an explicit contract: Agent authors must evaluate and declare compatibility with sub-agent constraints (e.g., no dependency on writing to parent memory, no reliance on cross-scope events), avoiding hard-to-debug behavioral anomalies at runtime.

We need to evaluate what capability restrictions sub-agents have, to confirm whether this marker is necessary.

---

To Be Refined

• Timeout / cancellation semantics: call() currently does not provide timeout or cancellation. Further investigation of requirements is needed to refine the design.
• Sub-agent prompt injection and dispatch chain for ChatModel: The prompt injection format for exposing sub-agents as tools to the LLM, and the tool_call_action dispatch chain, need further clarification.
• External agent support framework: Consider providing SubagentSetup implementation helpers for gRPC, A2A (Agent-to-Agent) protocol, etc., to reduce the cost of integrating external agents.
• RpcOperator-based RemoteAgent design: The remote execution approach via RpcOperator (FLIP-577) on a separate ActionExecutionOperator is yet to be designed.

[joeyutong]

Thanks for the proposal. I understand the motivation for providing a unified invocation API for internal and external agents. However, I think the current Subagent abstraction mixes the agent being registered with the capability used to invoke it.

From the user's perspective, an external agent is a black-box invocation target, so call() naturally matches its execution boundary. An internal Agent, however, is already presented to users through actions, resources, and event-driven execution. Exposing the same Agent as a callable sub-agent makes users reason about two public execution models for the same abstraction: it is event-driven when used as a regular Agent, but request/response-oriented when used as a sub-agent.

This is not only an internal implementation difference. To understand the behavior of call(), users also need to know how the request becomes an input event, which output events form the result, and what marks the call as complete, such as quiescence. In other words, the callable view is exposed to users, while the semantic adaptation that creates this view remains implicit.

Therefore, the proposal unifies the invocation shape, but does not fully unify the user-facing semantics. From this perspective, the shared abstraction seems closer to a SubagentInvoker:

• for an external agent, it delegates to the invocation protocol provided by the external integration;
• for an internal agent, it invokes the Agent through a CallProtocol that defines request, result, and completion mappings.

Under this model, the authoring experience remains different where it should be. For an external agent, its integration can provide a custom SubagentInvoker that encapsulates the agent endpoint and its invocation protocol. For an internal agent, users continue to define and register the Agent itself as ResourceType.AGENT; the framework provides a built-in SubagentInvoker, invisible to users, that drives the Agent according to the internal call protocol. Users therefore do not construct or register this invoker as another kind of Agent: SubagentInvoker is the invocation mechanism, while Agent remains the user-facing authoring abstraction.

This would preserve a unified API for callers without exposing call() as a second intrinsic execution model of every Agent.

[pltbkd]

Hi joeyutong,

Thanks for sharing your thoughts. I understand the confusion. The InternalSubagent preview in the proposal intentionally simplified many of the details, so let me briefly clarify a few points here.

First, I'd like to clarify that Subagent is a separate interface intended for callers and is independent of the concrete Agent implementation. Callers always use subagent.call(), regardless of whether the target is an internal or external subagent. However, only external subagents need to implement this method themselves; for internal subagents, the framework handles it transparently. In that sense, its role is essentially the same as the SubagentInvoker abstraction you described.

Any Agent that satisfies the required constraints can be registered as an internal subagent without any modification. InternalSubagent is a framework-provided abstraction that translates a call() request into an InputEvent within the subagent scope and returns the subagent's OutputEvent as the call result. Once the event is emitted, the calling Action yields, and the remainder of the execution continues under the existing event-driven execution model. There will be some framework changes to ensure that SubagentCallEvent can be dispatched and processed at that point.

In short, callers only need to understand subagent.call() and do not need to care whether the target is internal or external. Implementers only need to develop a standard Agent—they do not need to modify it or understand or implement call(). From the framework's perspective, there is still only a single event-driven execution model; subagent.call() is simply a resource-oriented invocation interface built on top of that model.

I hope this addresses your concerns. If anything is still unclear, I'd be happy to discuss it further here, or we can dive into the details in the dedicated InternalSubagent discussion.

[joeyutong]

Thanks for the clarification. I think we are aligned on the implementation architecture. What you describe as InternalSubagent is essentially the framework-provided implementation of the internal call protocol that I mentioned above. It also clarifies the distinction between the underlying Agent, which users author, and the caller-side invocation capability, which the framework provides for an internal agent.

Subagent is the common interface for all sub-agents. Both external (SubagentSetup) and internal sub-agents implement it.

My remaining concern is how this distinction is represented in the public API. While call() is a natural contract for an external integration, the proposal applies the same public Subagent abstraction to internal agents. An internal Agent author does not need to implement call(), but call() still becomes part of how the internal subagent is publicly represented, even though it is a framework-provided capability rather than part of the Agent authoring contract.

Therefore, my concern is not with the InternalSubagent implementation. This is why I characterized the shared interface as being closer to SubagentInvoker: not necessarily to prescribe a particular solution, but to emphasize that, for an internal sub-agent, the interface describes a framework-provided invocation capability rather than part of the Agent authoring contract.

[ofekron]

The SubagentInvoker / CallProtocol split above also gives durable execution a place to define a critical external-agent contract.

Caching a completed Result handles replay after the local completion record exists. It does not cover the crash window where an external agent accepted or completed a side effect, but Flink failed before checkpointing the result. Re-running the HTTP callable can duplicate work.

I would require each invocation protocol to define:

• a stable invocation ID propagated as the remote idempotency key;
• accepted, running, completed, failed, and unknown/reconciling states;
• a probe/reconcile operation after recovery before any retry;
• cancellation semantics and whether cancellation is only requested or confirmed;
• an explicit completion predicate rather than inferring success from internal quiescence.

Internal agents can implement that protocol through scoped events; external integrations can map it to HTTP, A2A, or another transport. The caller still receives one SubagentInvoker API, but durability claims remain honest per protocol. If an integration cannot reconcile an unknown outcome, fail closed and surface it instead of automatically repeating the call.

We use the same distinction in Better Agent when recovering native Claude, Codex, and Gemini runs: durable run identity and reconciliation are separate from the provider transport. I maintain the project; it is source-available and free for non-commercial use, while commercial use requires separate permission: https://github.com/ofekron/better-agent

AI-assistance disclosure: this comment was drafted by Codex under the maintainer’s authorization and reviewed in Better Agent.

[pltbkd]

Thanks for the detailed thoughts. This design aims to keep the user-facing interface simple. I agree that these concerns belong to the framework rather than every individual subagent implementation. At the same time, I'd also like to clarify that most of the mechanisms you mentioned are already part of the current design, although they are exposed differently:

• Stable Invocation ID → the deterministic SessionId.
• Invocation lifecycle → all subagent calls are expected to go through durableExecute.
• Recovery / reconciliation → handled by durableExecute, not by the subagent author.
• Completion semantics → durableExecute provides durable execution for internal subagents. Combined with framework-managed Action state, it enables strict consistency; without Action state, it naturally degrades to completion-only semantics, where intermediate execution state is not observable. External subagents are expected to provide the same completion semantics by using durableExecute.

Cancellation semantics are indeed missing today. I suppose it won't changes the overall architecture, but it is definitely a gap that should be addressed.

For external subagents, the intention is that they follow the same durability contract (for example, by wrapping external operations with durableExecute). What is currently missing is a shared framework for external subagents that can enforce these conventions automatically rather than relying on implementations to follow them. If you have concrete requirements or examples in this area, I'd be happy to discuss them further.