yunfengzhou-hub opened a new pull request, #1138:
URL: https://github.com/apache/flink-agents/pull/1138
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 isolated
memory and a child `ResourceCache` inheriting the root.
5. **Accumulate + resume + recover.** Child `OutputEvent`s 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),
scope-aware context, and isolated memory; 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`
|
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, lifecycle,
continuation 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.
<details>
<summary>Implementation invariants, full test roster, and change
size</summary>
- `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`, `ActionTaskContextManagerTest`,
`ContinuationSupportTest`, `TaskLifecycleListenerNotificationTest`; Python
`test_agent_plan_subagent_resources.py`, `test_task_lifecycle_bridge.py`; e2e
`internal_subagent_test.py`.
- Change size: 39 files, +2963/−66 across `api`, `plan`, `runtime` (Java),
and `python`.
</details>
### 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
- [x] `doc-needed`
- [ ] `doc-not-needed`
- [ ] `doc-included`
### Was this patch authored or co-authored using generative AI tooling?
- [x] Yes
- [ ] No
Generated-by: Qoder 1.29.0 (Qwen3.8-Max)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
