wenjin272 commented on PR #1220:
URL: https://github.com/apache/flink-agents/pull/1220#issuecomment-6119334624
I've updated this PR so that chat calls and internal subagents share an
internal subgraph runtime, rather than maintaining two orchestration paths or
implementing chat as a subagent.
### Architecture
```text
ctx.chat Internal subagent submit
| |
Chat adapter Subagent adapter
+-------------------+--------------------+
|
Shared subgraph invocation runtime
|
Existing Event / Action execution
```
The shared layer has three main responsibilities:
- `SubgraphDefinition<R>` defines private event-to-action routing,
identifies output events, and validates/assembles the result when the graph
finishes.
- `SubgraphInvocation<R>` tracks one call: its environment, collected
outputs, unfinished actions and result future. An action remains in flight
across async yield/resume. Its completion is counted only after its durable
outputs have been dispatched, so receiving an output event alone does not
finish the call.
- `SubgraphCallManager` is the operator-wide invocation registry and cleanup
mechanism. It is not another scheduler: `ActionExecutionOperator` still creates
and runs all action tasks.
`SubgraphEvent` carries an invocation ID and the underlying business event.
The operator routes it through the corresponding definition instead of the
outer graph's listeners; actions receive the unwrapped business event. Only
explicitly marked bootstrap events can be dispatched while the caller is
suspended. Ordinary events remain buffered as before.
### What stays in each adapter
| | Chat | Internal subagent |
|---|---|---|
| Graph | Existing Chat/Tool actions and their loop | Actions and conditions
from the child Agent plan |
| Input/output | `ChatRequestEvent` -> exactly one matching
`ChatResponseEvent` | `InputEvent` -> collected `OutputEvent` payloads |
| Resources and memory | Caller resources and shared memory stores, with
separate action journals | Child resource environment and isolated per-call
memory |
| Public result | `DurableFuture<ChatResult>` | Existing subagent handles /
`SubagentResult` |
`ChatInvocation` and `InternalSubagentCallStatus` both extend
`SubgraphInvocation`. Their adapters retain input/output conversion and
existing durable result boundaries; there is no extra durable wrapper around
the shared future. `RunnerContextImpl` holds the environment of the currently
executing action, and `ActionTaskContextManager` restores it on task switches.
A chat inside a subagent therefore uses the child environment but its own
private Chat/Tool routing.
### Execution and recovery
On first await, a chat call either reuses its recorded terminal response or
registers an invocation and emits its bootstrap. The caller yields, the
existing Chat/Tool actions run, and the shared invocation completes after all
scheduled actions retire. The chat adapter records the terminal response and
resumes the caller.
The invocation future and action counters are runtime-only, not
checkpointed. Recovery starts from root callers and reconstructs the call tree.
Completed child `ActionState` entries replay their outputs; incomplete actions
re-enter execution and reconcile durable operations. Completed callers do not
replay their consumed child bootstraps. This avoids independently restoring
both a caller and its child graph.
Captured subagent action failures are persisted as `SubgraphFailureEvent`
outputs containing diagnostic type/message text. Live execution and replay both
reconstruct the failure, rather than allowing a failed child to become an empty
successful graph after recovery. A failed graph waits for already scheduled
siblings to retire and suppresses their successor dispatch. Chat's ordinary
business failures remain FAILED `ChatResponseEvent`s and raise the existing
chat exception on await. Cancellation, fatal errors, and failures outside the
action-body boundary still propagate.
### Main integration and scope
This revision is rebased on main, including #1185: `ctx.chat` returns the
complete `ChatResult`, preserving the assistant message and available model,
response ID, usage, finish reason and metadata. The Java/Python tests now use
typed tool-call/result blocks, and result metadata is checked across JSON and
durable replay.
No public subgraph DSL or standalone Tool/RAG call API is added. Java awaits
still require the JDK 21 continuation runtime; the shared-worker limitation
(#1213) and unified caller identity work (#1084) remain separate. Migration of
in-flight internal-event state from the old implementation is not supported.
Validation on the rebased revision: Java API/Plan/Runtime/MCP on JDK 21
(2,439 passed, 44 skipped), JDK 11 focused regression (173 passed, 20 skipped),
Python API/Plan/Runtime (1,217 passed, 15 skipped), and Flink 2.3 MiniCluster
tests with rebuilt artifacts (4 Java tests on JDK 21 and 5 Python tests on JDK
11 passed). The PR body includes the contract-to-test mapping and remaining
verification gaps.
--
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]
