pltbkd opened a new pull request, #1299:
URL: https://github.com/apache/flink-agents/pull/1299
Linked issue: #1213
### Purpose of change
The core goal is to support subgraph execution on the non-coroutine parallel
engine, with two derived secondary goals: converge chat and sub-agent execution
onto subgraph, and improve the parallel engine's lifecycle semantics.
#### Why
A caller must release operator ownership for its child graph to progress.
Sharing the call protocol lets continuations and execution-lock handoff provide
that capability without separate chat and sub-agent schedulers. Explicit
suspend/resume keeps shared contexts and listeners aligned with ownership
changes.
#### Runtime flow
`SubgraphExecution` registers a call and reads its durable terminal slot:
hit → reuse; miss → bootstrap. Java drives `await`/`finish`; Python yields
between `tryFinish` polls. Terminal consumption follows retirement of scheduled
child actions and records only new results. The Python internal-call bridge
returns null while pending, or a JSON terminal envelope decoded into Python
`SubagentResult`.
The parallel engine dispatches bootstrap before the caller releases its lock
to wait. Dependent tasks bypass root FIFO execution/commit ordering. Resume
restores contexts before preparation callbacks. Checkpoint drain stops new root
pulls but allows outstanding/dependent work to commit; recovery rebuilds
dependent work from roots and reuses available child state.
#### Key decisions
- Share dispatch/state/recovery; `SubgraphDefinition` retains routing,
result, scope and failure policy. Internal calls are not deferred calls.
- Use action-owned IDs. Only required bootstrap may dispatch early; ordinary
outputs wait for action completion.
- Keep one pool: core one, ceiling `max(10, 2 * num-async-threads)`.
Headroom is not a general deadlock solution; input topology still limits safe
depth/fan-out.
- Defer serial/parallel chain convergence and system-wide exception
classification. Historical state/semantic compatibility is not required.
### Behavioral Semantics
#### Interaction decisions
| Caller / runtime | No recorded terminal result | Recorded terminal result |
|---|---|---|
| Java, JDK 21+ continuation | Yield continuation; graph wait holds no
worker | Return recorded result; no bootstrap |
| Java, JDK < 21 parallel, all-Java plan | Dispatch bootstrap, wait on
caller worker, hand off lock | Return recorded result; no bootstrap |
| Java, JDK < 21 without parallel engine | Reject before bootstrap | Same
capability guard rejects |
| Python, serial or continuation runtime | Poll and yield on
interpreter/mailbox thread | Return recorded result without polling yield |
| Python action in parallel engine | Unsupported; non-Java child plans are
rejected at open | Not an alternate replay path |
On a recovery slot miss, the graph restarts and may reuse child state;
without durable state it recomputes. Captured internal-child failures wait for
scheduled siblings to retire. Raw chat-action failures fail the task, rather
than completing a failed response for the caller. Cross-action ordering and
exactly-once external effects are not promised.
#### Behavioral contracts
- **C1:** Java chat/internal calls, including nesting, work on the parallel
engine within available capacity.
- **C2:** Recorded terminal results skip child execution and are not
recorded again. Python internal-call results use the same JSON value
representation on fresh execution and replay.
- **C3:** Recovery reuses durable state or recomputes deterministic work;
sink delivery may repeat.
- **C4:** Graph events and synchronous chat responses stay private to the
call.
- **C5a:** Ordinary internal child-action exceptions become
`SubagentResult.error`.
- **C5b:** Raw chat-action failures fail the task; failed chat responses
raise to the caller.
- **C6:** Chat uses caller resources/memory; internal calls use child
resources and isolated sensory/short-term stores, not isolated long-term memory.
- **C7:** Generated IDs are deterministic per action/target and survive
transfer. Explicit session IDs remain caller-owned.
- **C8:** Callbacks are operator-serialized, not thread-affine. Preparation
enters a phase; transfer may retain the same task.
- **C9:** `onActionFailed` reports processing failure, not phase exit; it
need not follow a new preparation.
- **C10:** Record callbacks follow admission/retirement, including no-action
and resumed records; failure need not emit record finish.
- **C11:** Drain preserves outstanding work; a matching checkpoint abort
interrupts its wait, while an older abort is ignored.
#### Failure behavior
Unsupported Java waiting fails before bootstrap; non-Java child plans fail
parallel-operator open. Duplicate IDs, invalid graph lifecycle operations and
chat output counts other than one raise protocol errors. Existing provider/tool
timeout and business-error handling remain; no whole-subgraph timeout is added.
Runtime folding excludes cancellation, JVM errors and faults still marked
`UncapturableException`. Suspend failure retains the lock and skips
supplier/resume; resume failure is primary, suppressing any supplier failure.
Interrupted lock reacquisition preserves interruption and skips resume. Normal
listener failures may propagate; failure-notification fan-out collects them
without replacing the original error.
**Technical debt:** generic handlers can translate framework faults into
business results. Local shell guards are not an end-to-end guarantee against
persistence or a guarantee of failover. This system-wide classification gap is
not specific to subgraph/tool calls.
**Usage limits:** await internal and deferred handles in separate batches.
Normally omit session ID; explicit IDs require caller-guaranteed uniqueness.
Reuse a generated ID only for the same conversation where continuation is
supported; **internal sub-agents do not support session continuation**.
Finite-pool exhaustion remains possible. Drain has no independent deadline:
interruption of a stalled drain depends on Flink's matching checkpoint-abort
notification.
### Tests
Earlier verification, before the trace and JSON-bridge fixes: JDK 11 **217
passed / 12 skipped**; JDK 21 core **190 passed / 42 skipped**, adjacent suites
**99 passed / 14 skipped**; Python's six focused runtime suites **65 passed**.
These suites had no failures or errors, but were not all rerun after the fixes.
Focused verification of the reviewed fix snapshot: JDK 11 **43 passed / 7
skipped**; Python's seven focused runtime suites **71 passed**;
real-Pemja/PyFlink internal-subagent e2e **4 passed** (success, failure result,
nesting and structured payload). Spotless and Ruff pass. These are separate
runs, not cumulative totals or a full-suite verification of the subsequently
rewritten branch. Java tests use JDK-gated paths: a skipped parallel test on
JDK 21 is not parallel-engine coverage.
| Contract | Tests / evidence boundary |
|---|---|
| C1 | `SubgraphParallelExecutionTest`: chat, internal, nested, concurrent
and single-async-thread calls; JDK 21 `ChatCallDeadlockTest` |
| C2 | `ChatCallTest.terminalResponseReplaysWithoutBootstrap`;
`InternalSubagentBridgeTest.terminalResultReachesTheBridgeIdenticallyAfterAFailover`:
successful/failed terminal JSON survives an `ActionStateSerde` round trip,
with no bootstrap or second journal write (not a Flink checkpoint restart).
Python polling tests and JDK 11/21 native probes cover conversion; the four e2e
cases do not restore a checkpoint |
| C3 | `SubgraphParallelFailoverTest.recoversGraphCallAfterTaskFailure`
(chat/internal/nested success);
`SubgraphParallelExecutionTest.restoresAndContinuesAfterCheckpoint` |
| C4 | `SubgraphParallelExecutionTest` root leakage guard and positive
control; nested-scope privacy also relies on routing structure |
| C5a |
`SubgraphInvocationTest.failedActionOutcomeSurvivesActionStateRoundTrip`;
`InternalSubagentBridgeTest` failed-terminal journal round trip; real-Pemja
`test_python_internal_subagent_failure_surfaces_via_result`; no failing-child
Flink checkpoint replay test |
| C5b | `ChatCallTest` failed-response replay; no dedicated raw-chat-action
fault test |
| C6 | `ActionTaskContextManagerTest`, `SubgraphCallTest`,
`InternalSubagentMatchTest`; no fresh native cross-language isolation run |
| C7 | `DeterministicIdAllocatorTest`,
`ActionTaskContextManagerTest.idCountersRestoreActiveCallAndSurviveContinuationTransfer`,
`BaseSubagentSetupTest`, `DeferredSubagentSubmitTest` |
| C8 | `ContinuationActionExecutorTest`,
`TaskLifecycleListenerNotificationTest`, Python lifecycle/base-subagent suites |
| C9 | `TaskLifecycleListenerNotificationTest` failure paths and Python
lifecycle tests; no claim that failure reopens a phase |
| C10 | `TaskLifecycleListenerNotificationTest` no-action and resumed-record
cases; coverage is engine-dependent |
| C11 | `ParallelExecutionCoordinatorTest` dependent/drain cases,
`ActionExecutionOperatorDrainDeadlockIntegrationTest`, capacity integration
tests |
Local native Pemja probes on JDK 11 and 21 verified identical fresh/replayed
Python values for integer-key maps, nested JSON and null outputs, plus
empty-output and failed-result conversion. JDK 11 parallel trace probes
verified `_internal_subgraph` event records and their producer/upstream
metadata under both default event-only and all-record logging, without
malformed action records. These probes are not repository regression tests;
parallel trace coverage remains a repository test gap.
Coverage targets ownership handoff, bounded-capacity progress, terminal
replay, successful-call recovery and native Python bridge conversion. **Not
verified:** full Java/Python suites on the final tree; native Python
checkpoint-restart recovery or cross-language memory isolation; every
partial-journal or failing-sibling checkpoint window; unbounded/deep workload
liveness; exhaustive serial/parallel callback equivalence. The system-wide
exception-classification gap and unsupported mixed batches/session continuation
remain explicit limitations, not successful recovery claims.
<details>
<summary>Implementation invariants and focused verification details</summary>
- Mutations require mailbox/execution-lock ownership. Preparation reads the
ordered durable cursor once. Default IDs include the target; explicit session
IDs gain no implicit resource discriminator.
- Invocations count logical actions through durable-output dispatch.
Captured failures drop later sibling outputs but wait for sibling retirement.
Internal calls remain registered until record cleanup; chat calls are removed
at terminal consumption.
- Dependents remain outstanding through commit. Drain returns root permits
and resubmits them afterward, without parking workers. Executor tracking
retains only live threads.
- Parallel subgraph dispatch creates an event trace context for event
notification and child-action creation, preserving producer/upstream metadata
rather than reusing the caller's action trace context.
- Java/Python bindings retain identity/handle state across suspension. The
internal-call bridge serializes the terminal `SubagentResult` with the
journal's JSON mapper and preserves its success/result/error fields; Python
decodes that envelope rather than converting a native Java result list.
Earlier Java runs used JDK-specific `JAVA_HOME`, then `mvn --offline -pl
runtime -am -Dspotless.skip=true -Dsurefire.failIfNoSpecifiedTests=false
-Dtest=<selectors> test-compile jar:test-jar test`. An earlier JDK 11 rerun had
**39 passed** across `SubgraphParallelExecutionTest`,
`SubgraphParallelFailoverTest`, `TaskLifecycleListenerNotificationTest` and
`DeterministicIdAllocatorTest`, after rebuilding stale JDK 21 test classes.
The fix-focused Java run used JDK 11 and `mvn --offline -B
--no-transfer-progress -pl dist/common,dist/flink-2.2 -am -Dspotless.skip=true
-Dsurefire.failIfNoSpecifiedTests=false
-Dtest=SubgraphParallelExecutionTest,InternalSubagentBridgeTest,PythonBridgeContractTest,ChatCallTest,SubgraphInvocationTest,TaskLifecycleListenerNotificationTest
test-compile jar:test-jar package`. Both matching `ChatCallTest` classes are
included; the operator-level class accounts for the seven JDK-gated skips.
Python focused tests:
`test_{flink_runner_context,base_subagent,chat_call,internal_subagent,internal_subagent_descriptor,task_lifecycle_bridge,deferred_subagent}.py`,
via the repository virtual environment's `python -m pytest`, with its
site-packages on `PYTHONPATH`. The four e2e tests in
`internal_subagent_test.py` ran with Python 3.11, JDK 11 and PyFlink 2.2.0
against the reviewed Python source and freshly repackaged distribution JARs;
`-Dmaven.jar.forceCreation=true` prevented reuse of a stale shaded bridge
class. Format checks: `mvn --offline -pl runtime -am spotless:check`, Ruff
`check --no-fix` and `format --check`.
</details>
### API
Public call shapes, lazy submission and action-local handles remain;
submitted sub-agent handles must be resolved before action completion. No YAML
option is added. JDK < 21 Java chat requires the all-Java parallel engine; JDK
21 still requires `--add-exports=java.base/jdk.internal.vm=ALL-UNNAMED`.
Python internal-call payloads follow JSON value semantics from the first
result: object keys are strings, and nested containers/nulls are decoded as
Python JSON values. Arbitrary native Python object types are not preserved. The
Python-facing Java polling bridge returns a nullable JSON `String`; the
Java-native internal-call result path is unchanged by this normalization.
Internal inheritance, runtime bridges and lifecycle extension contracts
change: `InternalSubagentSetup` no longer derives from `DeferredSubagent`, and
chat-specific runtime managers are removed. There is no historical
identity/state migration guarantee. Unrelated serial/parallel differences
remain.
### Documentation
- [ ] `doc-needed`
- [ ] `doc-not-needed`
- [x] `doc-included`
### Was this patch authored or co-authored using generative AI tooling?
- [x] Yes
- [ ] No
Generated-by: Qoder
--
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]
