wenjin272 opened a new pull request, #1348:
URL: https://github.com/apache/flink-agents/pull/1348
Linked issue: #1338 (Java implementation; Python follow-up after this PR
merges).
### Purpose of change
**This PR changes Java implementation and tests only. Python will be aligned
in a follow-up after the Java changes are merged.** Java callers can gather
ordinary async/durable, Chat and subagent calls through one scheduler, also
used by individual awaits. ToolCallAction uses this path for mixed
tool/subagent batches.
#### Runtime flow
```text
durableExecuteAsync -> SingleDurableFuture
chat -> SubgraphFuture
subagent submit -> SubagentFutureImpl -> durable or subgraph delegate
|
single await / gather await
-> shared scheduler: validate -> prepare -> start -> finish -> close
|
ToolCallAction: ordinary Tool -> framework durable wrapper
AsyncTool -> callAsync -> mapped ToolResponse
-> one gather -> responses associated with original call IDs
```
#### Key decisions
- Keep AsyncFuture as the public await interface. SchedulableFuture supplies
runtime scheduling; GatherAsyncFuture only represents the batch result.
- Share SubgraphFuture between Chat and internal subagents.
SubagentFutureImpl retains identity/query/cancellation/cleanup and delegates
execution. MappedAsyncFuture converts results without another durable boundary.
- Keep Tool.call synchronous. Plan-layer AsyncTool accepts RunnerContext;
runtime SubagentTool owns submission/normalization, supplied through
ResourceContext.getSubagentTool.
- Move ResourceContext.fromGetResource to TestResourceContexts in the API
test-jar.
### Behavioral Semantics
#### Interaction decisions
| Call / entry | Execution and recovery |
| --- | --- |
| Ordinary callable, single or gathered | Submit work through the runtime
executor; cache locally without a durable slot. |
| Durable callable, single or gathered | Reserve a pending slot before
execution; replay terminal results, reconcile pending calls when supported,
otherwise rerun. |
| Chat / internal subagent | Use SubgraphFuture; graph waits do not occupy
async workers. Requires a continuation or parallel engine. |
| Already-submitted external subagent | Submission stays separate; gathering
waits for the existing remote call. |
| Tool call | Ordinary tools receive a framework durable wrapper; AsyncTool
supplies its own handle. The parallel path gathers calls. |
#### Behavioral contracts
1. Results retain input order; completed handles reuse their cached outcomes
on repeated awaits.
2. Individual await and gather share execution/recovery logic, but only
gather applies batch parallelism and timeout settings.
3. Matching terminal records replay without execution; pending calls remain
recoverable. Ordinary async calls are not persisted.
4. Duplicate calls, foreign-context handles, unsupported custom handles and
nested batches are rejected.
5. Cancellation/failure exits release pending subagent registrations.
ContinuationNativeCallable still executes on the calling thread.
6. Tool results retain their call IDs; subagent normalization, metrics
binding and reporting are preserved. Subagent-specific reporting names still
use the reserved namespace.
#### Failure behavior
Ordinary callable exceptions become Outcome failures in gather and are
rethrown by individual await. The scheduler propagates
cancellation/interruption, UncapturableException and fatal Errors without
persisting them. Timeout cancels unfinished work: started durable callables
record timeout outcomes; unstarted callables and incomplete graphs retain
pending records.
ToolCallAction converts non-cancellation Exceptions, including
preparation/normalization failures, to tool responses. Invalid/cyclic subagent
results become tool failures; cancellation and fatal Errors propagate.
Exception classification in #1322 remains separate.
### Tests
| Contract | Evidence |
| --- | --- |
| 1: ordering and cached awaits | MixedFutureGatherTest;
JavaRunnerContextImplDurableExecuteAsyncTest |
| 2: common lifecycle, single-call timeout isolation |
schedulesInterfaceImplementationWithoutAsyncFutureBaseClass;
singleAwaitDoesNotApplyBatchTimeout |
| 3: replay, pending recovery and ordinary calls |
partialRecoveryReplaysCompletedToolAndRestartsOnlyPendingGraph;
completedRecoveryStartsNeitherWorkersNorGraphs; InternalSubagentRecoveryTest |
| 4: input validation | duplicate/foreign/custom/nested-handle cases in
JavaRunnerContextImplDurableExecuteAsyncTest |
| 5: cleanup, cancellation and inline execution | SubagentAwaitCleanupTest;
cancellationNeverBecomesADurableOutcome;
testDurableExecuteAsyncContinuationNativeReconcilableRunsInline |
| 6: tool adaptation and reporting | ToolCallActionSubagentTest;
ToolCallActionReportTest |
Java regression: **JDK 21: 540 passed, 17 skipped; JDK 11: 496 passed, 51
skipped.** Includes local HTTP subagents and operator recovery/parallel tests.
JDK 11 Spotless passed.
Not verified: Python or Java/Python bridge parity, live model services, the
complete repository suite, JDK 17 execution, cross-version savepoint upgrades,
or execution of the merge result with latest main.
<details>
<summary>Implementation invariants and verification details</summary>
DurableCallSlot reserves journal positions during prepare in input order;
finish writes to the reserved index. Mapping shares underlying call identity
and does not allocate another slot. A call-ID mismatch truncates the recorded
suffix before reserving the new call. The scheduler fences queued work before
cancellation, preserves available completions at timeout, and closes
participants on every exit.
Tests ran on branch head 69da2fa29 with the reactor selection below; Java 11
used a clean rebuild to remove Java 21 test artifacts:
```text
mvn -pl runtime -am package
-Dtest=*Subagent*Test,*Subgraph*Test,Chat*Test,*RunnerContext*Test,DurableExecutionContextTest,MixedFutureGatherTest,AsyncBatchExecutorTest,ToolCallAction*Test
-Dsurefire.failIfNoSpecifiedTests=false -Dsurefire.useFile=true
```
Java 21 used spotless.skip=true and both runs used the Byte Buddy agent.
API/plan/runtime totals were 557 tests on JDK 21 and 547 on JDK 11, including
skips. The branch merges cleanly with main at 720a4ece4; this is a conflict
check, not a merged-tree test run.
</details>
### API
SubagentFuture implements AsyncFuture; DurableFuture inherits await.
ResourceContext implementations must provide getSubagentTool. The API
SubagentTool and public callable-prefix constant move to runtime/plan, and
fromGetResource moves to test utilities.
Tool.call and RunnerContext entry-point signatures stay unchanged. ctx.chat
still returns ChatResult and throws for failed ChatResponseEvent.
Single async durable calls now reserve pending state before execution,
including calls without a reconciler. On JDK 11, ordinary single async calls
use the same worker pool as gathered calls. Python follows Java merge.
### 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: Codex CLI 0.153.4 (GPT-6; exact model version unavailable)
--
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]
