[yunfengzhou-hub]

Linked issue: #1137

Purpose of change

Registering an Agent instance as an AGENT resource now runs it as an in-process internal sub-agent: the caller resolves it via ctx.get_resource(scope, AGENT), calls submit(ctx, prompt), and awaits a Result holding the child's accumulated output — reusing an existing agent as a child instead of hand-writing a DeferredSubagentSetup. The child runs its own action loop, isolated and durable, and may nest.

Runtime flow

1. Compile. AgentPlan splits an AGENT resource into three exclusive cases — SubagentSetup, YAML descriptor, Agent instance. Only the Agent case compiles, via InternalSubagentCompilationHelper: the child plan is compiled once, shared across names registering the same child, with cycles and self-references rejected, and served by InternalSubagentProvider. A Python-authored child travels as a PythonSerializableResourceProvider.
2. Materialize. The provider builds InternalSubagentSetup (Java extends BaseDeferredSubagentSetup, Python subclasses DeferredSubagentSetup) holding child plan and scope. Per-call statuses, per-scope child caches, and the per-key session index live in the setup, not the operator.
3. Submit + bootstrap. submit(ctx, prompt) assigns (session_id, call_id) from the executing task's namespace at await time, so a replay awaiting in the same order re-derives the same ids, and returns a deferred SubagentFuture. On first resolve prepare registers the call status and sends one InternalSubagentCallEvent on the mailbox thread, outside the durable boundary so a replayed send carries identical attributes.
4. Dispatch + suspend. The callable waits off the mailbox thread — a Java caller on a JDK 21 continuation, a Python coroutine at await. The operator matches the child's actions to the envelope and runs them scope-aware (ActionTaskContextManager binds the child's SubagentScope; child resources resolve against the child plan, including a nested sub-agent) with memory isolated for the whole call and a child ResourceCache inheriting the root's Python bridge and carrying the child plan JSON, so a Python-owned child resource materializes against its own plan.
5. Accumulate + resume + recover. Child OutputEvents accumulate into the call status; on quiesce the operator completes its future and the caller resumes with SubagentResult.ok. findCallStatus recurses through child caches so nested envelopes resolve depth-agnostically. On failover the replayed bootstrap reproduces the identity, so the child replays its recorded output; onRecordFinished(key) drops that key's sessions.

Key decisions

• Compile the child Agent rather than require a DeferredSubagentSetup subclass: that boilerplate is what this removes, and a real plan and action loop compose (nesting) where a single deferred callable cannot.
• Opaque deferred handle, not a raw future. Awaiting must release the mailbox so the operator can dispatch the child; a CompletableFuture.get() would block the mailbox and deadlock.
• Identity assigned at await, not minted at send, so failover re-derives it and the replayed envelope resolves the child's persisted state instead of re-running it.
• Orchestration state in the setup, not the operator, keeping per-scope and per-key state off the operator's shared dispatch path.

Behavioral Semantics

Interaction decisions

Caller context	On resolve
Java action, JDK 21 continuation	Caller suspends on the continuation; the mailbox dispatches the child; caller resumes with the Result.
Java action, no stackful suspension (≤ JDK 17)	prepare fails fast with IllegalStateException rather than blocking the mailbox and deadlocking.
Python async action	await releases the mailbox; a Python-compiled child's lazily-built handle adopts the prepared namespace, so no-id submit mints deterministic ids.
Python synchronous action	Holds the mailbox for its whole pemja call and cannot yield, so awaiting an internal sub-agent would deadlock; the caller action must be async.
Any caller, after failover	Replayed bootstrap carries identical attributes; the child replays its recorded output and ids re-derive identically.

Behavioral contracts

• An Agent registered as an AGENT resource is invocable in-process; awaiting submit(ctx, prompt) returns a Result whose result holds the child's accumulated OutputEvent payloads.
• A child action that raises surfaces as a failed Result (success=false, error_message); the job continues.
• A child may register and await its own sub-agents; nested envelopes resolve to the right depth.
• The child is isolated: its own resource cache (inheriting the root's Python bridge), scope-aware context, and memory kept for the whole call with parent MemoryItems copied before exposure, so a child write never mutates the parent's field list; its events do not land on the parent context.
• Successive calls under one record run independently under distinct identities.
• A completed call is durable: after failover it resolves to the same identity and replays the child's recorded output.

Failure behavior

• Resolve without stackful suspension (Java caller): raises IllegalStateException (fail-fast, no deadlock).
• Python prepare with a context that is not an InternalSubagentCallFactory: raises NotImplementedError.
• Bootstrap send fails, or await for an unregistered (sessionId, callId): raises IllegalStateException.
• No-id submit on a handle never wired to the task lifecycle: raises RuntimeError ("No prepared action task to assign sub-agent ids from"); the operator's Python-lifecycle gate keeps a Python caller driving a Python-compiled child from reaching it.
• Child action raises: captured into SubagentResult.error(e); the caller's Result reports failure and the job proceeds.
• Unresolved handles at action finish: the deferred base enforces resolution; a failed invocation skips cleanup because the run fails and replays.

Tests

Contract	Coverage
Agent compiles to an internal provider; shared plan; cycles/self-reference rejected; other two AGENT cases unaffected	childAgentCompilesIntoInternalProvider, sharedChildAgentCompilesToSinglePlan, cycleThroughRootIsRejected, selfReferenceIsRejected, nonSubagentAgentResourceIsRejected; Python test_agent_plan_subagent_resources.py mirrors all
Caller receives accumulated output	callerReceivesTheChildsAccumulatedOutput; e2e test_python_internal_subagent_call_end_to_end
Child failure surfaces via Result	childFailureSurfacesThroughTheResult; e2e test_python_internal_subagent_failure_surfaces_via_result
Nesting; scope matching	nestedCallReachesTheGrandchild, scopeMatchingAppliesTypeAndConditionFilters; e2e test_python_internal_subagent_nested_call
Independent identities per call	successiveCallsRunIndependentlyUnderDistinctIdentities
Fail-fast without continuations	resolvingWithoutContinuationsFailsFast, continuationsAreAvailableOnTheTestClasspath
Durability / failover replay	InternalSubagentRecoveryTest; replayOfCompletedActionEmitsPreparedAndReusedPair
Cross-language lifecycle and no-id namespace replay	TaskLifecycleListenerNotificationTest; Python test_task_lifecycle_bridge.py
Child memory persists for the whole call (a later child action reads an earlier one's write)	laterChildActionReadsValueWrittenByEarlierChildAction
Child field write does not mutate the parent's field list	childFieldWriteDoesNotLeakIntoParentFieldList, childNestedWriteDoesNotLeakIntoParentObject
Nested child plan survives Java↔Python round-trip	internalSubagentProviderSurvivesJsonRoundTrip, javaPlanWithInternalSubagentSnapshotIsStable; Python test_python_can_deserialize_java_plan_with_internal_subagent
Internal child's Python-owned resource materializes through the root bridge	testInternalChildCacheMaterializesPythonOwnedResourcesThroughTheRootBridge; Python test_scoped_materialization_uses_the_child_plan_and_its_own_cache, test_close_closes_the_root_and_every_scoped_cache

Highest-risk paths — cross-language identity replay, mailbox suspension/deadlock, and cycle rejection — are pinned by dedicated Java and Python tests plus a real-Flink Python e2e (call, failure, nested); plan-compilation parity is asserted in both languages.

Verified: Java ut-java green (internal sub-agent, plan, memory, lifecycle, continuation, resource-cache suites); Python non-integration suite green; internal sub-agent Python e2e green on a real Flink job; ./tools/build.sh --java on JDK 17.

Not verified: live-cluster checkpoint/failover of an in-flight call (recovery is harness-level, not a real cluster restart); a Java caller awaiting end-to-end through the JDK 21 continuation on a live cluster; external-service e2e. Residual fork CI failures are pre-existing LLM-auth (HTTP 401) integration tests, unrelated to this change.

Implementation invariants, full test roster, and change size

• prepare splits a mailbox-thread bootstrap (registers InternalSubagentCallStatus, indexes the record key, registers the session owner on the shared context, sends the envelope) from an off-mailbox call/_call that waits on the status's response future and converts a child exception into SubagentResult.error.
• awaitSubagentCall unwraps ExecutionException so the caller's Result carries the failing child action's exception, not future plumbing.
• Cross-language no-id: ActionExecutionOperator registers the PythonTaskLifecycleListener when the root plan declares a Python-compiled sub-agent (hasPythonCompiledSubagent()), so notify_action_prepared reaches FlinkRunnerContext, which captures the prepared namespace and replays it onto the lazily-materialized handle via adopt_prepared_namespace; SubagentIdAllocator is scoped to one action execution so ordinals restart per action and replays re-derive the same ids.
• Full roster: Java InternalSubagentCallTest/RecoveryTest/GuardTest/MatchTest, AgentPlanSubagentResourceTest, AgentPlanCrossLanguageTest, IsolatedCachedMemoryStoreTest, ResourceCacheTest, ActionTaskContextManagerTest, ContinuationSupportTest, TaskLifecycleListenerNotificationTest; Python test_agent_plan_subagent_resources.py, test_agent_plan_cross_language.py, test_eager_materialize.py, test_flink_runner_context_close.py, test_task_lifecycle_bridge.py; e2e internal_subagent_test.py.
• Change size: 57 files, +4447/−131 across api, plan, runtime (Java), and python.

API

No public signature changes. Agent.addResource(String, ResourceType, Object) is unchanged; for an AGENT resource it now also accepts an Agent instance (javadoc-documented widening, source-compatible) alongside the existing SubagentSetup and YAML descriptor forms, which are unaffected. Python submit(ctx, prompt) gains working no-id support for internal sub-agents; submit_with_identity remains the explicit-id hook. New plan and runtime classes are additive. A Java caller awaiting in-process requires JDK 21 (continuation); the Python caller path does not.

Documentation

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

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

• Yes
• No

Generated-by: Qoder 1.29.0 (Qwen3.8-Max)

[purushah]

Thanks for picking this up. Before we get into line-by-line review, could we open a discussion for the runtime model? This PR settles a few things #909 left open, and they affect most of the operator changes.

The three I'd like agreement on:

1. Where in-flight call state lives across a checkpoint. Right now it's only on the heap and rebuilt when the parent replays, but child tasks are checkpointed before the parent is re-enqueued. On restore the child can run first, find no status, and the job loops.
2. Whether the parent should wait on a ContinuationActionExecutor thread. Children use the same pool, so enough concurrent calls on one subtask can starve it.
3. What "done" means for a child with more than one action. Replay only re-sends the child's output events, so a chained child comes back with partial output after failover.

I have more detailed notes and am happy to share them once the direction is set.

[yunfengzhou-hub]

Thanks for picking this up. Before we get into line-by-line review, could we open a discussion for the runtime model? This PR settles a few things #909 left open, and they affect most of the operator changes.

The three I'd like agreement on:

1. Where in-flight call state lives across a checkpoint. Right now it's only on the heap and rebuilt when the parent replays, but child tasks are checkpointed before the parent is re-enqueued. On restore the child can run first, find no status, and the job loops.
2. Whether the parent should wait on a ContinuationActionExecutor thread. Children use the same pool, so enough concurrent calls on one subtask can starve it.
3. What "done" means for a child with more than one action. Replay only re-sends the child's output events, so a chained child comes back with partial output after failover.

I have more detailed notes and am happy to share them once the direction is set.

@purushah Thanks for the comment. Worth aligning on these before this moves further. The preview does have some known persistence gaps that we're addressing.

1. In-flight state and recovery. We'd like to settle the principle first and get your read. A sub-agent call inherits durable execution's consistency model:

• Without ActionState, an in-flight call tolerates recomputation and takes the final result.
• With ActionState, the call is consistent but requires the request to be replayable; there is no reconcile mechanism, and anything needing reconcile belongs in durable execution.

Mechanically, the parent's initiation and the child action's start each create a sub-agent call record, the later one linking to the earlier. On failover: without ActionState, in-flight calls are cancelled and re-run; with ActionState, the replayed action records serve as a reconciler for the call — when the parent has initiated but not completed, its request is matched against the persisted action records, completed actions are skipped and their results reused (including an already-produced OutputEvent, the sub-agent's response), and only the unfinished child actions re-run. Consistency therefore still rests on durable execution. I can write the full plan up separately if that's easier to react to.

2. Shared continuation executor. The call awaits on the async pool, so it does hold async threads; under heavy parallel fan-out where the children also need async capacity, this can block. We expect large fan-out to be rare, so the near-term guidance is to raise the async thread count, with auto-scaling as a later option.

3. Multi-action child "done". A call completes when all actions in its sub-agent scope quiesce — the same condition the main agent uses to finish a record: every output action is an OutputEvent and none is pending. Consistency follows point 1.

---

We'd like to continue discussions in this PR or #1137, given that after #909 we have created an umbrella issue for sub-agent mechanism to better track discussions and development in each child issue. What do you think?

[pltbkd]

I have a proposal for the recovery consistency model.

Given that:

1. Internal sub-agent recovery can provide the same recovery strength as durable execution under deterministic replay.
2. ActionState serves an action-level role analogous to reconciliation: one record is initialized before execution, updated as durable calls finish, and marked completed with the action's memory updates and emitted events when the action finishes.
3. The deterministic attributes of InternalSubagentCallEvent, including the sub-agent identity, form the event-identity component of the ActionState key. Together with the same business key, sequence number, and action identity, replay can therefore locate the original ActionState entry after failover.

We do not need to restore child ActionTasks. We can retain only unfinished root callers and let them issue the calls again. The regenerated child tasks then reconcile against ActionState: an incomplete action reruns and reuses its persisted durable-call results, while a completed action skips its body and replays the framework-visible events needed to reconstruct the remaining action graph. Calls to a deeper-level sub-agent initiated by that completed action can be skipped because their results have already been consumed by the action. Without ActionState, the unfinished subtree is recomputed.

This makes the recovery strategy clear and simple. I have sent the code to @yunfengzhou-hub for review. Do you think this strategy is feasible?

[wenjin272]

Thanks for implementing this and documenting the runtime flow! I left four inline comments on memory handling and cross-language resource support.

The in-flight recovery ordering, multi-action replay, and shared async-pool starvation concerns already raised by @purushah also remain on this head. I reproduced the two recovery cases and confirmed the shared-pool dependency in the code, so I have not duplicated those inline.

[wenjin272]

Could we retain child memory for the lifetime of the sub-agent call? persistCache() is called after each action, but clearing the cache here discards writes before the next child action can read them. I reproduced this with action A writing a value and emitting an event to action B: B could no longer read the value.

[yunfengzhou-hub]

Fixed. The child's isolated memory now persists for the whole call instead of being cleared after each action, so a later child action reads what an earlier one wrote. laterChildActionReadsValueWrittenByEarlierChildAction reproduces your A-writes/B-reads case.

[wenjin272]

Nesting an AgentPlan here exposes a missing final writeEndObject() in AgentPlanJsonSerializer: the provider type marker ends up inside childPlan, so the resulting JSON fails deserialization in both Java and Python. After correcting the nesting, Python still fails because its plan deserializer does not handle InternalSubagentProvider. Could we fix both sides and add Java/Python round-trip tests?

[yunfengzhou-hub]

Both sides fixed. Java adds the missing final writeEndObject() so the provider type marker no longer leaks into childPlan; Python's deserializer now converts InternalSubagentProvider into InternalSubagentSetup. Round-trip coverage: internalSubagentProviderSurvivesJsonRoundTrip (Java→Java), javaPlanWithInternalSubagentSnapshotIsStable (Java→snapshot), and test_python_can_deserialize_java_plan_with_internal_subagent (Java→Python).

[wenjin272]

Could we copy the mutable MemoryItem before exposing it to the child? MemoryObjectImpl.set() mutates the returned item's subKeys in place, so a child write changes the parent's field list even though the value stays in the child cache. I reproduced a parent getFields() NPE after the child added a field.

[yunfengzhou-hub]

Fixed. The parent MemoryItem is copied with an independent subKeys set before being exposed to the child scope, so a child field write no longer mutates the parent's field list. childFieldWriteDoesNotLeakIntoParentFieldList and childNestedWriteDoesNotLeakIntoParentObject cover it.

[wenjin272]

This child cache never receives a PythonActionExecutor or PythonResourceAdapter; the parent reference does not inherit those fields. Consequently, an internal child registering a Python-owned external sub-agent fails during eager materialization, even when the root Python runtime is initialized. Could we wire the bridge into child caches and add coverage for this nested resource case?

[yunfengzhou-hub]

Fixed. The child cache now inherits the Python bridge from the root (effectivePythonActionExecutor/effectivePythonResourceAdapter fall back to the parent) and carries the child plan JSON, so a Python-owned child resource materializes against its own plan through the single root runtime. Coverage: testInternalChildCacheMaterializesPythonOwnedResourcesThroughTheRootBridge (Java) and test_scoped_materialization_uses_the_child_plan_and_its_own_cache (Python).

[yunfengzhou-hub]

I have a proposal for the recovery consistency model.

Given that:

1. Internal sub-agent recovery can provide the same recovery strength as durable execution under deterministic replay.
2. ActionState serves an action-level role analogous to reconciliation: one record is initialized before execution, updated as durable calls finish, and marked completed with the action's memory updates and emitted events when the action finishes.
3. The deterministic attributes of InternalSubagentCallEvent, including the sub-agent identity, form the event-identity component of the ActionState key. Together with the same business key, sequence number, and action identity, replay can therefore locate the original ActionState entry after failover.

We do not need to restore child ActionTasks. We can retain only unfinished root callers and let them issue the calls again. The regenerated child tasks then reconcile against ActionState: an incomplete action reruns and reuses its persisted durable-call results, while a completed action skips its body and replays the framework-visible events needed to reconstruct the remaining action graph. Calls to a deeper-level sub-agent initiated by that completed action can be skipped because their results have already been consumed by the action. Without ActionState, the unfinished subtree is recomputed.

This makes the recovery strategy clear and simple. I have sent the code to @yunfengzhou-hub for review. Do you think this strategy is feasible?

@pltbkd Feasible — adopted and implemented(co-authored to you, now pushed to the PR). It follows your model faithfully: drop child tasks on resume, replay from the root caller, and reconcile completed actions against their persisted ActionState. It also closes @purushah 's concern #3 (partial output after failover) as a side effect — a yield now holds same-call and ordinary events until the action finishes, so ActionState captures them and replay reconstructs the full output instead of losing mid-yield events.

[wenjin272]

Thanks for the follow-up fixes and regression tests! The original checkpoint-ordering, chained-replay, and cross-action memory reproducers now pass. I left five follow-up comments on remaining recovery, resource ownership, Python initialization, and memory isolation issues, including a concrete inline follow-up to the shared-pool starvation concern already raised by @purushah.

To clarify priorities:

• The failure on a second recovery and shared async-pool starvation are P1 issues that I think should be addressed before merging.
• The two cross-language resource issues (internal setups being overwritten during mixed registration, and Python dependencies being missed when they exist only in child plans) are P2, non-blocking follow-ups. These should not block this PR.
• The mutable memory-value isolation issue is also P2.

[wenjin272]

Could we also exclude sub-agent bootstraps when replaying a completed root action? After a child result is persisted and the parent fails, recovery still runs prepare(), but the cached result returns without yielding. The bootstrap then gets persisted among the parent's output events. On another recovery, this branch replays it without registering its call status, causing Missing subagent call status. I reproduced this with two successive recovery attempts.

[yunfengzhou-hub]

Fixed. The non-subagent branch of completedActionOutputEvents() now drops InternalSubagentCallEvent and replays only the graph-driving events, so a completed root caller no longer re-dispatches its finished child's bootstrap envelope — which registers no call status and caused the Missing subagent call status abort.
Added InternalSubagentRecoveryTest#replaysCompletedRootCallerWithoutRebootstrappingItsFinishedChildCall, an operator-level snapshot→restore test that seeds a completed root caller whose persisted outputs still hold the child bootstrap envelope, then restores and asserts the child is not re-dispatched.

[wenjin272]

Java and Python disagree on resource ownership here: Java excludes internal child-plan providers from Python-owned resources, while Python's is_python_owned() still includes them. When internal and external Python sub-agents coexist, Python eager materialization returns both, and this loop replaces the existing Java InternalSubagentSetup with a PythonRuntimeResource. The subsequent internal call fails its type check. Could we align ownership classification and cover this mixed-registration case?

[yunfengzhou-hub]

Fixed. Python's is_python_owned() now excludes internal child-plan providers, matching Java's isPythonOwned, so a Python-compiled internal sub-agent stays Java-owned and eager materialization no longer replaces the InternalSubagentSetup with a PythonRuntimeResource. Added a mixed internal + external Python sub-agent registration case to test_resource_provider.py.

[wenjin272]

Inheriting the root bridge only works if the root initialized Python. PythonBridgeManager.open() still checks only the root plan's actions and resources. With a Java root whose child contains the only Python resource, no Python runtime is initialized, so child materialization still fails with "no Python runtime was initialized." Could we discover Python dependencies recursively across child plans? I reproduced this through the actual initialization and child-cache paths.

[yunfengzhou-hub]

Fixed. PythonBridgeManager.open() now detects Python across the whole plan tree: a cycle-guarded planTree() DFS feeds treeContainsPythonAction() / treeContainsPythonResource(), so a Java root whose only Python signal sits in a child plan still initializes the runtime. Mem0 detection stays root-only. Added PythonBridgeManagerTest#planTreeRecursesThroughNestedInternalSubagentChildPlans (detection) and #openStartsPythonWhenOnlyASubagentChildPlanBearsPython (open() actually enters the init branch on the child's signal).

[wenjin272]

Copying subKeys fixes the field-list corruption, but value is still shared with the parent. I reproduced a child reading a parent-owned list, modifying it, and writing it back to the child memory: the parent's list also changed. Could we isolate mutable payloads as well, so lists and maps cannot bypass the child memory boundary?

[yunfengzhou-hub]

After discussion with @pltbkd , MemoryItem is now immutable and child scopes isolate by copy-on-write, so a mutable payload (list/map) read from the parent is copied into the child cache before any write and can no longer mutate the parent in place. This supersedes the earlier subKeys-only copy. Covered by MemoryObjectTest and IsolatedCachedMemoryStoreTest.

[wenjin272]

Thanks for the update. I reran the mutable-list reproducer on 7dca9eb, and the parent value still changes. IsolatedCachedMemoryStore.get() returns the parent's item, and MemoryItem.getValue() exposes the same payload reference. Making that reference final does not make the list/map immutable:

List<String> items = (List<String>) child.get("items").getValue();
items.add("child"); // Already mutates the parent's list, before child.set().

The copy-on-write change protects subKeys, but this path bypasses it. Could we also isolate mutable payloads and add a regression test asserting that modifying a list obtained through child memory leaves the parent's value unchanged?

[yunfengzhou-hub]

The deep copy of MemoryItem.getValue() is a little more complicated than expected, especially for external subagent. After discussion with @pltbkd , I think it might be better to isolate memory between parent and child for now, so I've just reverted changed related to memory sharing ability.

[wenjin272]

Could we suspend the parent wait without occupying a worker in the shared async pool? awaitSubagentCall() blocks a thread that the child may also need for its durable async operations. With num-async-threads=1, a single parent awaiting a child that performs an async call can deadlock; larger pools hit the same condition with enough concurrent or nested calls. Increasing the thread count raises the threshold but does not remove this dependency.

[yunfengzhou-hub]

Agreed — this one is broader than a scoped fix: the parent's await holds a worker from the same bounded async pool its child needs, so num-async-threads=1 can self-deadlock and a larger pool only raises the threshold. I've filed #1213 to track suspending the wait without occupying a pool worker. I'd suggest keeping this PR to the four correctness fixes above and taking the threading change in #1213 — does that work for you?

[wenjin272]

Thanks for addressing the comments. LGTM