[api][runtime][java][python] Add event-backed chat calls from actions

[wenjin272]

Linked issue: #1086

Purpose of change

Actions can use Java ctx.chat(...).await() or Python await ctx.chat(...) without separate request/response actions. Scope: chat and its tool rounds, not standalone Tool/RAG calls.

Runtime flow

1. ctx.chat snapshots the request and returns a lazy DurableFuture<ChatResult>.
2. First await reuses a recorded response or registers a ChatInvocation and queues a private SubgraphEvent bootstrap. The caller suspends while existing Chat/Tool actions run.
3. Chat and subagents share a subgraph runtime. The existing operator schedules actions; adapters supply routing, resource/memory scopes and result contracts.
4. After all scheduled actions retire, chat validates exactly one matching response. The caller records it when durability is enabled, then returns the complete result or raises the chat error.
5. Recovery reconstructs child calls from root callers and reconciles child ActionState; runtime futures are not checkpointed.

Key decisions

• Reuse event orchestration for routing, retries and tool rounds. Chat is a subgraph adapter, not a subagent wrapper. No public subgraph API is added.
• Keep existing chat/subagent durable-result boundaries; do not add a second durable slot for the shared future.
• Release only internal bootstraps while suspended. Ordinary events retain their existing ordering.

Behavioral Semantics

Interaction decisions

Conditions	Result
Java + JDK 21 continuation context	Caller yields; an async worker waits. Unsupported contexts reject before dispatch.
Python async caller + JDK 11 or 21	Coroutine yields without a response-wait worker.
Durable execution + recorded terminal response	Replay success/failure without another chat bootstrap.
No recorded result / durability disabled	Run the event flow; incomplete external effects may repeat on recovery.
Chat inside an internal subagent	Inherit child resources and caller memory stores; retain private Chat/Tool routing.
Subagent failure + scheduled siblings	Suppress successor dispatch and wait for siblings before completing the failed call.
Java wait + async model/tool execution	Share the worker pool; insufficient capacity can deadlock (#1213).

Behavioral contracts

1. Unawaited chat futures dispatch nothing. Repeated awaits within the creating action execution reuse the outcome.
2. Await returns ChatResult, including message blocks and available invocation metadata; FAILED responses raise a catchable chat exception.
3. Private chat events do not trigger outer listeners. Ordinary events stay buffered until the calling action finishes.
4. Chat shares caller resources and memory stores; internal subagents retain child resources and isolated per-call memory.
5. Durable replay reuses recorded outcomes and reconstructs unfinished graphs from root callers. This does not guarantee exactly-once external side effects.
6. Captured subagent failures remain failures after recovery. Results wait for all scheduled actions to retire.
7. Chat futures cannot be composed with gather; cross-action use is unsupported and unchecked. Existing model/tool timeouts apply; no whole-chat timeout is added.

Failure behavior

FAILED chat responses raise ChatResponseEvent.ChatResponseException (Java) or ChatResponseError (Python). Missing, duplicate or mismatched responses are protocol errors. Captured subagent action exceptions persist as diagnostic type/message events, reconstructed as IllegalStateException and exposed through the subagent error result. Cancellation, fatal errors, serialization, persistence and dispatch failures propagate. Invalid requests and unsupported runtime/composition raise without fallback.

Tests

Contract	Coverage
Lazy/repeated awaits; full result and metadata (1, 2)	ChatDurableFutureTest, ChatCallManagerTest, Python test_chat_call.py
Isolation and ordinary-event ordering (3)	ChatCallTest, SubgraphCallTest, Java/Python MiniCluster chat tests
Caller/child environments (4)	ChatCallTest.chatInsideInternalSubagentUsesChildResources, Python nested-chat and internal-subagent E2E
Replay and checkpoint recovery (5)	ChatCallManagerTest, ChatCallTest, InternalSubagentRecoveryTest
Failure replay and graph completion (6)	SubgraphInvocationTest, InternalSubagentRecoveryTest, SubgraphCallTest
Unsupported composition and runtime (7)	Java/Python gather-rejection and Java continuation-guard tests; no cross-action misuse guard/test or dedicated whole-call timeout test

Validation for this 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.
• Clean Java build and rebuilt/installed wheel; Flink 2.3 MiniCluster: 4 Java tests passed on JDK 21; 5 Python tests passed on JDK 11.
• Spotless, Ruff and git diff --check passed.

Not verified: live-provider/routing/structured-output/media E2E, real-MiniCluster chat recovery (operator-harness coverage only), deadlock resolution, state migration, or E2E beyond Flink 2.3 in this revision.

Implementation invariants and supporting coverage

• SubgraphCallManager is a mailbox-confined registry, not a scheduler. Invocation action counts survive yield/resume and retire only after durable output dispatch. Fan-out, empty graphs, nested-call cleanup and failure siblings have focused tests.
• SubgraphEvent isolates routing while preserving delegate identity for ActionState lookup. Completed callers do not replay consumed bootstraps. Recovery tests cover before dispatch, after child completion, after await, nested trees, partial outputs and captured failures.
• ChatCallOwner retains the existing caller-identity allocator; unified identity work ([Bug][runtime] Durable action-state keys collide for distinct events with identical attributes #1084) is not solved here.
• Deterministic MiniCluster providers use typed ToolCallBlock/ToolResultBlock data, not live external services. Java JARs and the Python wheel are rebuilt before E2E.

API

Add Java RunnerContext.chat(ChatRequestEvent) / chat(String, List<ChatMessage>) and Python RunnerContext.chat(model, messages, *, prompt_args=None, output_schema=None), returning DurableFuture<ChatResult>. Event APIs, YAML syntax and subagent public APIs are unchanged. Java awaits require JDK 21 continuation support. Captured subagent failures reconstruct diagnostic exceptions; the original exception object is not preserved. Old in-flight internal-event state cannot be migrated.

Documentation

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

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

• Yes
• No

Generated-by: Codex CLI 0.153.4 (GPT-5)

[wenjin272]

Please review this PR when you get a chance @pltbkd.

[sangkyoonnam]

Two events with equal attributes from one input record can make overlapping executions of the same action fail at ctx.chat, even with durable execution off (inline, ChatCallManager.java:134). Otherwise FAILED-response replay, the cancellation and persistence error paths, and the gather rejection held up. Local checks passed: 268 runtime tests (23 skipped), 521 Python API and chat-call tests, and 344 Python runtime tests; I didn't run the MiniCluster tests.

I also probed a crash after the tool ran but before the second model turn: recovery gave tool=1, model=2 and one output, because the bootstrap request's id is deterministic and goes into the envelope attributes that child action-state lookup uses. The existing phase 3 stops after model call 2, so a test for that mid-round point seems worth adding.

[sangkyoonnam]

Two in-flight callers with the same identity collide here. ChatCallOwner derives the id from SubagentIdAllocator (key, sequence number, action name, event type and attributes, but not the event id), and ChatCallManager is operator-wide. In a probe where one action sends new Event("ask") twice and an ask action awaits ctx.chat(...) with two async threads, the second caller fails the task:

java.lang.IllegalStateException: Duplicate chat call: c56f87c4-9c56-3867-98c7-2d38b559b136-0

The probe configures no action-state store, so the older durable-key aliasing isn't what fails here; with a store enabled, identical events also alias through ActionStateUtil.generateUUIDForEvent, which hashes attributes only, and that's a separate recovery issue. Could these overlapping callers get distinct ids that stay replay-stable, with an operator test that runs with persistence off?

[wenjin272]

Thanks for catching this. The collision comes from a shared identity problem: internal subagents and chat calls use SubagentIdAllocator, while ActionStateUtil has the related issue tracked in #1084. Neither reliably distinguishes separate event occurrences with identical attributes.

I suggest addressing this through a unified, replay-stable execution identity shared by durable state, internal subagents, and chat calls, rather than adding a chat-specific fix in this PR. The chat case should be included in that work, with regression coverage for overlapping calls when persistence is disabled. cc @pltbkd

Would you be okay with tracking this under #1084 and keeping the fix separate from this PR?

[sangkyoonnam]

Sounds good, I'm fine tracking this under #1084. I reran the same probe on a5e6270, and it now fails in the shared registry with the same id: Duplicate subgraph call: c56f87c4-9c56-3867-98c7-2d38b559b136-0. Chat calls and internal subagents both register through SubgraphCallManager now and both still derive their ids through SubagentIdAllocator, so a fix that only covers durable-state keys would leave this persistence-off case open. I can post the chat repro on #1084 so whoever picks it up has that case to test against.

[sangkyoonnam]

Within one caller identity, replay matches the next slot by builtin-chat:<id>, and that id carries the await ordinal but no digest of the model, messages or prompt arguments. In a probe that recorded a response for "hello", stopped at a later durableExecuteAsync, changed the prompt to "changed" and restored, the action got answer:hello back with one model call. That follows the stable-id-first contract durable_execute uses, so this is non-blocking and docs-only: chat_models.md:275 could say replay goes by caller identity and await order without comparing request contents, so changing a request at the same position reuses the recorded response.

[pltbkd]

Hi @wenjin272, thanks for this PR. I went through it and found it is using some components of the internal sub-agent.

I suppose it's because the root requirement is a common one: dispatch a child event flow, suspend the caller, and resume it on a terminal event. Internal sub-agent also requires isolation, but most of the execution is the same, and both face the same issues like triggering and restoring.

I'd suggest a full refactor of the internal sub-agent execution in this PR, separating the execution part from the isolation part so the former can be shared between chat and sub-agent instead of duplicating the implementation and the problem solving; with chat re-implemented on that basis, its own implementation should stay thin enough to keep the overall change size comparable to this PR. I'm willing to help with the development and the review. The determinism question can be discussed separately — the current trade-off matches how internal sub-agent and action state behave today, where two events with identical attributes are not supported yet.

What do you think?

[wenjin272]

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

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 ChatResponseEvents 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.