weiqingy commented on code in PR #938:
URL: https://github.com/apache/flink-agents/pull/938#discussion_r3746302853
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -403,6 +440,7 @@ private void processActionTaskForKey(Object key) throws
Exception {
boolean currentInputEventFinished = false;
if (isFinished) {
+ notifyTaskFinished(actionTask);
Review Comment:
This fires after the action is already recorded as done.
`maybePersistTaskResult` has called `actionState.markCompleted()`
(`DurableExecutionManager.java:269`), and the output events went downstream at
line 437. So `checkEmpty` throwing here can't undo anything.
On the restart, the action passes `actionState.isCompleted()` at line 379
and gets skipped, so no handle is ever registered and the check quietly passes.
The dropped call ends up skipped either way, just one job failure later.
Would moving the check earlier help, right after `actionTask.invoke` returns
at line 415?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/subagent/BaseAsyncSubagentSetup.java:
##########
@@ -0,0 +1,340 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.subagent;
+
+import org.apache.flink.agents.api.context.DurableCallable;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.subagent.Result;
+import org.apache.flink.agents.api.subagent.SubagentFuture;
+
+import javax.annotation.Nullable;
+
+import java.util.concurrent.Callable;
+
+/**
+ * Production base for sub-agents whose protocol is an asynchronous job, run
in pub/sub mode: {@code
+ * submit} (the pub) starts the run remotely through one durable POST and
immediately returns a
+ * handle carrying the {@code (sessionId, callId)} identity; {@code isDone},
{@code await} and
+ * {@code cancel} on the handle (the sub) query or steer that run. The shape
matches LangGraph runs,
+ * OpenAI Assistants runs, and A2A long-running tasks.
+ *
+ * <h2>Integration primitives</h2>
+ *
+ * <p>Integrations only provide the transport primitives they already
understand, with no durable
+ * concepts involved:
+ *
+ * <ul>
+ * <li>{@link #callSubmitRequest} — start the run remotely; a thrown
exception fails the action;
+ * <li>{@link #callQueryStatus} — a read-only probe of the run's current
state;
+ * <li>{@link #callFetchResult} — fetch the result of a run that reached a
terminal state;
+ * <li>{@link #callCancelRequest} — optional hook propagating a cancellation
to the remote run.
+ * </ul>
+ *
+ * <h2>Persistence conventions</h2>
+ *
+ * <p>The framework wrappers decide which operation runs through durable
execution:
+ *
+ * <ul>
+ * <li>{@link #submitRequest} — durable, id {@code sessionId#callId}, the
only wrapper wired to a
+ * reconciler ({@link #reconcileSubmitRequest}), so the remote run is
started at most once
+ * even across a crash between the POST landing and its result being
persisted;
+ * <li>{@link #queryStatus} — not durable: a direct read-only probe on the
mailbox thread. The
+ * state advances monotonically toward a terminal state, so a replay
observing a fresher state
+ * is harmless;
+ * <li>{@link #fetchResult} — durable, id {@code sessionId#callId#fetch}:
the result enters the
+ * caller's data flow and must replay deterministically. No reconciler;
recovery re-executes
+ * the fetch, which is an idempotent read;
+ * <li>the await composition of {@code await} — durable, id {@code
sessionId#callId#await}: poll
+ * the status until a terminal state, then fetch;
+ * <li>{@link #cancelRequest} — not durable: a direct, synchronous
propagation. Remote
+ * cancellations are expected to be idempotent, so a replay propagating
the cancellation again
+ * is harmless.
+ * </ul>
+ *
+ * <p>The fetch and await ids are fixed per identity: both compositions are
built from idempotent
+ * reads, so a recovery re-executing them converges to the same outcome as the
original run.
+ *
+ * <h2>Cancellation contract (dev-facing)</h2>
+ *
+ * <p>Cancel decisions typically depend on nondeterministic inputs such as
processing time. A
+ * failover replay therefore does not promise control flow equivalent to the
original execution: the
+ * original may have taken a cancel branch that the replay skips, or vice
versa. The only
+ * at-most-once guarantee is the POST, enforced by the reconciler;
cancellation propagation is
+ * best-effort and idempotent. The hook returns nothing: a cancelled {@code
await} always fails as a
+ * {@link java.util.concurrent.CancellationException}, and a hook failure
propagates from {@code
+ * cancel} and fails the action.
+ *
+ * <h2>Known limitations</h2>
+ *
+ * <ul>
+ * <li>If the remote session or run record expires after a failover, the
non-durable {@link
+ * #queryStatus} may report a different state than before the crash, and
a replay may not be
+ * able to reproduce the original fetch path; persisted fetch records
still short-circuit;
+ * <li>If a fetch was in flight when the process crashed and the remote
fetch is consume-once
+ * rather than an idempotent read, the recovery re-execution cannot
recover the result — a
+ * reconciler cannot fix this; the remote protocol must guarantee
idempotent reads;
+ * <li>Any cancellation governs the subsequent {@code await} and may discard
a fetch that had
+ * actually succeeded, even when its durable record exists — cancel is
the authoritative
+ * control-flow decision.
+ * </ul>
+ */
+public abstract class BaseAsyncSubagentSetup extends BaseSubagentSetup {
+
+ /** Delay between status probes while waiting for the run to reach a
terminal state. */
+ protected long statusPollIntervalMillis = 10;
+
+ //
------------------------------------------------------------------------------------------
+ // pub: submit starts the run immediately through one durable POST
+ //
------------------------------------------------------------------------------------------
+
+ /**
+ * Starts the remote run through the durable POST and returns its handle.
A POST failure throws
+ * and fails the action.
+ */
+ @Override
+ public final SubagentFuture submit(
+ RunnerContext ctx, Object prompt, String sessionId, String callId)
throws Exception {
+ ctx.durableExecuteAsync(submitRequest(ctx, sessionId, callId, prompt));
+ return new AsyncSubagentFuture(this, ctx, sessionId, callId,
currentTaskRegistry());
+ }
+
+ //
------------------------------------------------------------------------------------------
+ // Framework wrappers: defaults composing the primitives, overridable
+ //
------------------------------------------------------------------------------------------
+
+ /** The durable POST of one invocation; the only wrapper wired to a
reconciler. */
+ protected DurableCallable<Void> submitRequest(
+ RunnerContext ctx, String sessionId, String callId, Object prompt)
{
+ return new DurableCallable<Void>() {
+ @Override
+ public String getId() {
+ return sessionId + "#" + callId;
+ }
+
+ @Override
+ public Class<Void> getResultClass() {
+ return Void.class;
+ }
+
+ @Override
+ public Void call() throws Exception {
+ callSubmitRequest(sessionId, callId, prompt);
+ return null;
+ }
+
+ @Override
+ public Callable<Void> reconciler() {
+ // Recovery never assumes the POST was lost: probe first,
resend only a missing
+ // run, so a crash after the POST landed never duplicates the
prompt.
+ return () -> {
+ reconcileSubmitRequest(sessionId, callId, prompt);
+ return null;
+ };
+ }
+ };
+ }
+
+ /** The status probe; not durable, a direct read-only query on the mailbox
thread. */
+ protected RunStatus queryStatus(String sessionId, String callId) throws
Exception {
+ return callQueryStatus(sessionId, callId);
+ }
+
+ /** The durable fetch of a terminal run's result, keyed by {@code
sessionId#callId#fetch}. */
+ protected DurableCallable<Result> fetchResult(
Review Comment:
Nothing in production calls this. `awaitResult` goes straight to
`callFetchResult` at line 207, so the `#fetch` slot described at line 60 never
gets written. The only caller is `fetchResultForTest`
(`MockAsyncSubagentSetup.java:155`), and Python is the same shape:
`fetch_result` (`async_subagent.py:286`) is test-only, and
`_await_until_terminal` calls `call_fetch_result` at line 322.
That also means the note at 86-89 about persisted fetch records
short-circuiting can't happen, since nothing writes one.
I can see why it went this way, `awaitResult.call()` runs off the mailbox so
it can't call back into `durableExecuteAsync`. Is the plan to make the fetch
slot reachable later, or would dropping it for now be cleaner?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/subagent/BaseAsyncSubagentSetup.java:
##########
@@ -0,0 +1,340 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.subagent;
+
+import org.apache.flink.agents.api.context.DurableCallable;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.subagent.Result;
+import org.apache.flink.agents.api.subagent.SubagentFuture;
+
+import javax.annotation.Nullable;
+
+import java.util.concurrent.Callable;
+
+/**
+ * Production base for sub-agents whose protocol is an asynchronous job, run
in pub/sub mode: {@code
+ * submit} (the pub) starts the run remotely through one durable POST and
immediately returns a
+ * handle carrying the {@code (sessionId, callId)} identity; {@code isDone},
{@code await} and
+ * {@code cancel} on the handle (the sub) query or steer that run. The shape
matches LangGraph runs,
+ * OpenAI Assistants runs, and A2A long-running tasks.
+ *
+ * <h2>Integration primitives</h2>
+ *
+ * <p>Integrations only provide the transport primitives they already
understand, with no durable
+ * concepts involved:
+ *
+ * <ul>
+ * <li>{@link #callSubmitRequest} — start the run remotely; a thrown
exception fails the action;
+ * <li>{@link #callQueryStatus} — a read-only probe of the run's current
state;
+ * <li>{@link #callFetchResult} — fetch the result of a run that reached a
terminal state;
+ * <li>{@link #callCancelRequest} — optional hook propagating a cancellation
to the remote run.
+ * </ul>
+ *
+ * <h2>Persistence conventions</h2>
+ *
+ * <p>The framework wrappers decide which operation runs through durable
execution:
+ *
+ * <ul>
+ * <li>{@link #submitRequest} — durable, id {@code sessionId#callId}, the
only wrapper wired to a
+ * reconciler ({@link #reconcileSubmitRequest}), so the remote run is
started at most once
+ * even across a crash between the POST landing and its result being
persisted;
+ * <li>{@link #queryStatus} — not durable: a direct read-only probe on the
mailbox thread. The
+ * state advances monotonically toward a terminal state, so a replay
observing a fresher state
+ * is harmless;
+ * <li>{@link #fetchResult} — durable, id {@code sessionId#callId#fetch}:
the result enters the
+ * caller's data flow and must replay deterministically. No reconciler;
recovery re-executes
+ * the fetch, which is an idempotent read;
+ * <li>the await composition of {@code await} — durable, id {@code
sessionId#callId#await}: poll
+ * the status until a terminal state, then fetch;
+ * <li>{@link #cancelRequest} — not durable: a direct, synchronous
propagation. Remote
+ * cancellations are expected to be idempotent, so a replay propagating
the cancellation again
+ * is harmless.
+ * </ul>
+ *
+ * <p>The fetch and await ids are fixed per identity: both compositions are
built from idempotent
+ * reads, so a recovery re-executing them converges to the same outcome as the
original run.
+ *
+ * <h2>Cancellation contract (dev-facing)</h2>
+ *
+ * <p>Cancel decisions typically depend on nondeterministic inputs such as
processing time. A
+ * failover replay therefore does not promise control flow equivalent to the
original execution: the
+ * original may have taken a cancel branch that the replay skips, or vice
versa. The only
+ * at-most-once guarantee is the POST, enforced by the reconciler;
cancellation propagation is
+ * best-effort and idempotent. The hook returns nothing: a cancelled {@code
await} always fails as a
+ * {@link java.util.concurrent.CancellationException}, and a hook failure
propagates from {@code
+ * cancel} and fails the action.
+ *
+ * <h2>Known limitations</h2>
+ *
+ * <ul>
+ * <li>If the remote session or run record expires after a failover, the
non-durable {@link
+ * #queryStatus} may report a different state than before the crash, and
a replay may not be
+ * able to reproduce the original fetch path; persisted fetch records
still short-circuit;
+ * <li>If a fetch was in flight when the process crashed and the remote
fetch is consume-once
+ * rather than an idempotent read, the recovery re-execution cannot
recover the result — a
+ * reconciler cannot fix this; the remote protocol must guarantee
idempotent reads;
+ * <li>Any cancellation governs the subsequent {@code await} and may discard
a fetch that had
+ * actually succeeded, even when its durable record exists — cancel is
the authoritative
+ * control-flow decision.
+ * </ul>
+ */
+public abstract class BaseAsyncSubagentSetup extends BaseSubagentSetup {
+
+ /** Delay between status probes while waiting for the run to reach a
terminal state. */
+ protected long statusPollIntervalMillis = 10;
Review Comment:
nit: 10ms works out to roughly 100 status calls a second per run, against
the services named at lines 34-35 (LangGraph, OpenAI Assistants, A2A), which
don't usually finish that fast. Tests set it to 0
(`MockAsyncSubagentSetup.java:71`), so integrations inherit this value. Would a
bigger default, or a backoff, fit better?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/subagent/SubagentFutureGroup.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.subagent;
+
+import org.apache.flink.agents.api.subagent.Result;
+import org.apache.flink.agents.api.subagent.SubagentFuture;
+import org.apache.flink.agents.api.subagent.SubagentFutures;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Batched resolve of several handles in submission order. The group knows
deferred handles: {@link
+ * #awaitAll} prepares every pending deferred handle up front, executes the
prepared calls as a
+ * batch, and only then collects the outcomes — so the requests are issued
together instead of one
+ * at a time as each wait starts. Already resolved handles simply contribute
their value.
+ */
+final class SubagentFutureGroup extends SubagentFutures {
+
+ private final List<SubagentFuture> futures;
+
+ SubagentFutureGroup(SubagentFuture first, SubagentFuture[] others) {
+ this(withFirst(first, others));
+ }
+
+ private static List<SubagentFuture> withFirst(SubagentFuture first,
SubagentFuture[] others) {
+ List<SubagentFuture> all = new ArrayList<>(1 + others.length);
+ all.add(first);
+ all.addAll(Arrays.asList(others));
+ return all;
+ }
+
+ private SubagentFutureGroup(List<SubagentFuture> futures) {
+ this.futures = futures;
+ }
+
+ @Override
+ public boolean isDone() {
+ for (SubagentFuture future : futures) {
+ if (!future.isDone()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public List<Result> awaitAll() throws Exception {
+ // Prepare every pending deferred handle up front, so the whole batch
is ready before any
+ // execution starts.
+ for (SubagentFuture future : futures) {
+ if (future instanceof DeferredSubagentFuture && !future.isDone()) {
+ ((DeferredSubagentFuture) future).prepare();
+ }
+ }
+ // TODO: execute the prepared calls as one batch once durable
execution supports batched
Review Comment:
Three different descriptions of the same behavior. This TODO says serial,
the class javadoc at 30-33 says the calls go out together as a batch, and the
public `SubagentFutures.awaitAll` javadoc says each handle resolves when its
own wait starts (`SubagentFutures.java:44`). Callers read that last one.
On the batching itself, #926 adds `reservePendingBatch` on both
`RunnerContextImpl` and the Python bridge, which looks like the piece this TODO
is waiting for. It's still open so nothing can lean on it yet, but have you
tried the two together? Curious whether preparing everything up front and
handing over one list fits, or whether the deferred mode needs something #926
doesn't expose.
##########
python/flink_agents/runtime/async_subagent.py:
##########
@@ -0,0 +1,396 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""The async-job execution mode running in durable pub/sub mode."""
+
+import time
+from abc import ABC, abstractmethod
+from concurrent.futures import CancelledError
+from enum import Enum
+from typing import Any
+
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.subagent import (
+ Result,
+ SubagentFuture,
+ SubagentFutures,
+)
+from flink_agents.runtime.base_subagent import BaseSubagentSetup
+from flink_agents.runtime.subagent_handles import SubagentFutureGroup
+
+
+class RunStatus:
+ """State snapshot of a remote run reported by the ``call_query_status``
+ probe.
+
+ A state other than ``NOT_STARTED`` means the submission landed on the
+ service, which is the sole basis for ``reconcile_submit_request``
+ deciding between re-posting and polling. The snapshot never carries the
+ result payload.
+ """
+
+ class State(Enum):
+ """Lifecycle of the remote run."""
+
+ NOT_STARTED = "not_started"
+ RUNNING = "running"
+ COMPLETED = "completed"
+ FAILED = "failed"
+
+ def __init__(self, state: "RunStatus.State", error: str | None = None) ->
None:
+ """Initialize with the lifecycle state and the optional error."""
+ self._state = state
+ self._error = error
+
+ @staticmethod
+ def not_started() -> "RunStatus":
+ """The service has no record of the run: the POST never landed."""
+ return RunStatus(RunStatus.State.NOT_STARTED)
+
+ @staticmethod
+ def running() -> "RunStatus":
+ """The run is in progress."""
+ return RunStatus(RunStatus.State.RUNNING)
+
+ @staticmethod
+ def completed() -> "RunStatus":
+ """The run finished successfully."""
+ return RunStatus(RunStatus.State.COMPLETED)
+
+ @staticmethod
+ def failed(error: str) -> "RunStatus":
+ """The run failed, carrying the error message."""
+ return RunStatus(RunStatus.State.FAILED, error)
+
+ @property
+ def state(self) -> "RunStatus.State":
+ """The lifecycle state of the remote run."""
+ return self._state
+
+ @property
+ def error(self) -> str | None:
+ """The error message of a failed run; None otherwise."""
+ return self._error
+
+
+class AsyncSubagentFuture(SubagentFuture):
+ """The sub side of an async-job invocation.
+
+ The run was already started by the durable POST of ``submit``, so the
+ handle only subscribes to it: :meth:`done` probes the status directly
+ (not durable); awaiting the handle waits through the durable await
+ composition; :meth:`cancel` propagates the cancellation through the
+ setup's hook, and a cancelled resolve raises ``CancelledError``.
+ """
+
+ def __init__(
+ self,
+ setup: "BaseAsyncSubagentSetup",
+ ctx: RunnerContext,
+ session_id: str,
+ call_id: str,
+ ) -> None:
+ """Initialize with the owning setup, the context, and the identity."""
+ super().__init__(session_id, call_id)
+ self._setup = setup
+ self._ctx = ctx
+ self._consumed = False
+ self._cancelled = False
+ self._value: Result | None = None
+
+ def done(self) -> bool:
+ """Probe the remote status directly; not durable.
+
+ A failover replay may probe a different number of times than the
+ original execution.
+ """
+ if self._consumed or self._cancelled:
+ return True
+ try:
+ probe = self._setup.query_status(self.session_id, self.call_id)
+ except Exception:
+ return False
+ return probe.state in (
+ RunStatus.State.COMPLETED,
+ RunStatus.State.FAILED,
+ )
+
+ def __await__(self) -> Any:
+ """Wait for the run through the durable await composition, releasing
+ the mailbox while waiting.
+
+ A cancelled handle raises :class:`CancelledError`.
+ """
+ if self._cancelled:
+ msg = f"Sub-agent call cancelled: {self.identity}"
+ raise CancelledError(msg)
+ if not self._consumed:
+ self._value = yield from self._ctx.durable_execute_async(
+ self._setup._await_until_terminal,
+ self.session_id,
+ self.call_id,
+ durable_id=f"{self.identity}#await",
+ ).__await__()
+ self._consumed = True
+ return self._value
+
+ def cancel(self) -> None:
+ """Propagate the cancellation through the setup's hook; not durable.
+
+ A failover replay (which creates a fresh handle) may propagate it
+ again; remote cancellations must be idempotent. A repeated cancel on
+ the same handle and a cancel after the resolve are local no-ops. A
+ cancelled resolve raises :class:`CancelledError`; a hook failure
+ propagates and fails the action.
+ """
+ if self._consumed or self._cancelled:
+ return
+ self._setup.cancel_request(self._ctx, self.session_id, self.call_id)
+ self._cancelled = True
+
+ def combine(self, *others: SubagentFuture) -> SubagentFutures:
+ """Group this handle with others for a batched resolve."""
+ return SubagentFutureGroup((self, *others))
+
+
+class BaseAsyncSubagentSetup(BaseSubagentSetup, ABC):
+ """Runtime base for sub-agents whose protocol is an asynchronous job,
+ run in pub/sub mode.
+
+ ``submit`` (the pub) starts the run remotely through one durable POST and
+ immediately returns a handle carrying the ``(session_id, call_id)``
+ identity; ``done``, the resolve and ``cancel`` on the handle (the sub)
+ query or steer that run. The shape matches LangGraph runs, OpenAI
+ Assistants runs, and A2A long-running tasks.
+
+ Integrations only provide the transport primitives they already
+ understand, with no durable concepts involved:
+
+ * :meth:`call_submit_request` — start the run remotely; a raised
+ exception fails the action;
+ * :meth:`call_query_status` — a read-only probe of the run's current
+ state;
+ * :meth:`call_fetch_result` — fetch the result of a run that reached a
+ terminal state;
+ * :meth:`call_cancel_request` — optional hook propagating a cancellation
+ to the remote run.
+
+ Persistence conventions:
+
+ * the submit POST — durable, id ``session_id#call_id``, the only
+ operation wired to a reconciler (:meth:`reconcile_submit_request`), so
+ the remote run is started at most once even across a crash between the
+ POST landing and its completion being persisted;
+ * :meth:`query_status` — not durable: a direct read-only probe. The state
+ advances monotonically toward a terminal state, so a replay observing a
+ fresher state is harmless;
+ * the fetch — durable, id ``session_id#call_id#fetch``: the result
+ enters the caller's data flow and must replay deterministically. No
+ reconciler; recovery re-executes the fetch, which is an idempotent
+ read;
+ * the await composition of the resolve — durable, id
+ ``session_id#call_id#await``: poll the status until a terminal
+ state, then fetch;
+ * :meth:`cancel_request` — not durable: a direct, synchronous
+ propagation. Remote cancellations are expected to be idempotent, so a
+ replay propagating the cancellation again is harmless.
+
+ The fetch and await ids are fixed per identity: both compositions are
+ built from idempotent reads, so a recovery re-executing them converges
+ to the same outcome as the original run.
+
+ Cancellation contract (dev-facing): cancel decisions typically depend on
+ nondeterministic inputs such as processing time, so a failover replay
+ does not promise control flow equivalent to the original execution. The
+ only at-most-once guarantee is the POST, enforced by the reconciler;
+ cancellation propagation is best-effort and idempotent. The hook returns
+ nothing: a cancelled resolve always raises ``CancelledError``, and a
+ hook failure propagates from :meth:`cancel` and fails the action.
+
+ Known limitations: a remote session or run record expiring after a
+ failover may change the non-durable status probe and prevent a replay
+ from reproducing the original fetch path (persisted fetch records still
+ short-circuit); a fetch in flight when the process crashed cannot be
+ recovered from a consume-once remote (the remote protocol must guarantee
+ idempotent reads); and any cancellation may discard a fetch that had
+ actually succeeded — cancel is the authoritative control-flow decision.
+ """
+
+ #: Delay between status probes while waiting for the run to reach a
+ #: terminal state; the Python parity of Java's statusPollIntervalMillis.
+ status_poll_interval_seconds: float = 0.01
+
+ def submit_with_identity(
+ self,
+ ctx: RunnerContext,
+ prompt: Any,
+ session_id: str,
+ call_id: str,
+ ) -> SubagentFuture:
+ """Start the remote run through the durable POST and return its
+ handle.
+
+ The POST runs on the mailbox thread and lands before the handle is
+ returned; a POST failure raises and fails the action.
+ """
+ self.submit_request(ctx, session_id, call_id, prompt)
+ return AsyncSubagentFuture(self, ctx, session_id, call_id)
+
+ #
--------------------------------------------------------------------------------
+ # Framework wrappers: defaults composing the primitives, overridable
+ #
--------------------------------------------------------------------------------
+
+ def submit_request(
+ self,
+ ctx: RunnerContext,
+ session_id: str,
+ call_id: str,
+ prompt: Any,
+ ) -> None:
+ """Run the durable POST of one invocation; the only wrapper wired to
+ a reconciler.
+
+ Recovery never assumes the POST was lost: the reconciler probes
+ first and resends only a missing run, so a crash after the POST
+ landed never duplicates the prompt.
+ """
+ ctx.durable_execute(
Review Comment:
This one is synchronous. `durable_execute` is documented as blocking the
operator until it finishes (`flink_runner_context.py:656`), so the whole remote
POST sits on the mailbox thread and nothing else moves while it's out.
Java does it async (`BaseAsyncSubagentSetup.java:113`), and the await path
in this same file already uses `durable_execute_async` at line 142.
Is the sync call intentional here, or should this match the await path?
##########
api/src/main/java/org/apache/flink/agents/api/subagent/SubagentFuture.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.api.subagent;
+
+/**
+ * Driving handle for one sub-agent invocation, carrying the {@code
(sessionId, callId)} identity
+ * that keys the invocation's durable state.
+ *
+ * <p>The handle is a driver of the invocation's lifecycle, not a passive
view: resolving it issues
+ * the request when it has not been issued yet, and its wait releases the
mailbox so other work can
+ * proceed in between. Its heap state does not survive a failover; the
identity is the only basis
+ * for rebuilding it through replay.
+ *
+ * <p>Returned by {@link Subagent#submit}. The invocation is always deferred:
the request is issued
Review Comment:
This says the request always goes out when the handle resolves, but the
async mode POSTs during `submit` (`BaseAsyncSubagentSetup.java:113`), which its
own javadoc calls the pub side.
It shows up in `cancel`: cancelling a deferred handle sends nothing,
cancelling an async one fires a remote cancel. Someone holding a
`SubagentFuture` can't tell which they have from this text.
Could this be scoped to the deferred mode?
##########
api/src/main/java/org/apache/flink/agents/api/subagent/Subagent.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.api.subagent;
+
+import org.apache.flink.agents.api.context.RunnerContext;
+
+/**
+ * Caller-facing interface for all sub-agents (external and internal).
+ *
+ * <p>An invocation is identified by a {@code (sessionId, callId)} pair; the
session groups a
+ * conversation across invocations. Callers do not manage ids: the short forms
below leave the
+ * missing ids to the implementation, which assigns them (runtime setups
typically through a
+ * deterministic id allocator, stable across failover replays) or rejects the
call.
+ *
+ * <p>The full form taking the complete {@code (sessionId, callId)} identity
is the
+ * implementation-side contract, declared by {@link SubagentSetup}; resolving
a returned handle is
+ * {@code await}.
+ */
+public interface Subagent {
+
+ /**
+ * Issues an invocation under the given {@code sessionId}; the
implementation picks the call id.
+ */
+ SubagentFuture submit(RunnerContext ctx, Object prompt, String sessionId)
throws Exception;
Review Comment:
nit: the PR description still describes the old surface. `asAsyncCallable`,
`callAsync` and `executeAllAsync` have no occurrences left in the branch, and
`call()` is gone from this interface. Worth refreshing it so the body matches
`cc114aef`?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/subagent/AsyncSubagentFuture.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.subagent;
+
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.subagent.Result;
+import org.apache.flink.agents.api.subagent.SubagentFuture;
+import org.apache.flink.agents.api.subagent.SubagentFutures;
+import
org.apache.flink.agents.runtime.subagent.BaseAsyncSubagentSetup.RunStatus;
+
+import javax.annotation.Nullable;
+
+import java.util.concurrent.CancellationException;
+
+/**
+ * The sub side of an async-job invocation: the run was already started by the
durable POST of
+ * {@code submit}, so the handle only subscribes to it. {@link #isDone()}
probes the status directly
+ * (not durable); {@link #await()} waits through the durable await
composition; {@link #cancel()}
+ * propagates the cancellation through the setup's hook, and a cancelled
{@code await} fails as a
+ * {@link CancellationException}.
+ *
+ * <p>The handle records itself in the owning base's per-task registry, so
dropping it without
+ * collecting the outcome fails the action instead of silently losing the
result.
+ */
+final class AsyncSubagentFuture extends SubagentFuture {
+
+ private final BaseAsyncSubagentSetup setup;
+ private final RunnerContext ctx;
+ @Nullable private final PendingSubagentCallRegistry registry;
+
+ private boolean consumed;
+ private boolean cancelled;
+ @Nullable private Result value;
+
+ AsyncSubagentFuture(
+ BaseAsyncSubagentSetup setup,
+ RunnerContext ctx,
+ String sessionId,
+ String callId,
+ @Nullable PendingSubagentCallRegistry registry) {
+ super(sessionId, callId);
+ this.setup = setup;
+ this.ctx = ctx;
+ this.registry = registry;
+ if (registry != null) {
+ registry.trackPendingSubagentCall(identity());
+ }
+ }
+
+ /**
+ * Probes the remote status directly; not durable, so a failover replay
may probe a different
+ * number of times than the original execution.
+ */
+ @Override
+ public boolean isDone() {
+ if (consumed || cancelled) {
+ return true;
+ }
+ try {
+ RunStatus probe = setup.queryStatus(getSessionId(), getCallId());
+ return probe.getState() == RunStatus.State.COMPLETED
+ || probe.getState() == RunStatus.State.FAILED;
+ } catch (Exception e) {
Review Comment:
A failing probe turns into `false` with nothing logged. If `queryStatus`
keeps failing, say expired credentials or a dead endpoint, `isDone()` reports
"not done" forever and there's no clue why. `Result.error` does log on the same
kind of path (`Result.java:79`). Worth a debug log here?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/subagent/BaseDeferredSubagentSetup.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.subagent;
+
+import org.apache.flink.agents.api.context.DurableCallable;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.subagent.Result;
+import org.apache.flink.agents.api.subagent.SubagentFuture;
+
+/**
+ * Framework-level deferred execution mode for sub-agent setups: invocations
run through one async
+ * callable, issued lazily through a deferred handle. The mode is not tied to
external services; an
+ * internal sub-agent drives its child plan through the same shape.
+ *
+ * <p>{@link #submit} always returns a deferred handle built on {@link
DeferredSubagentFuture}: the
+ * request is prepared when the handle is resolved. Implementations only
provide the terminal {@link
+ * #prepare}, which supplies the {@link DurableCallable} running the
invocation; the handle feeds it
+ * to durable execution itself.
+ *
+ * <p>The short {@code submit} forms are inherited from {@link
BaseSubagentSetup}: the base assigns
+ * the missing ids deterministically from the executing task before the
deferred handle is created.
+ * The dropped-handle safety net is built in: the handle records itself in the
base's per-task
+ * registry, and a handle left unresolved when the action finishes fails the
action.
+ */
+public abstract class BaseDeferredSubagentSetup extends BaseSubagentSetup {
+
+ @Override
+ public SubagentFuture submit(RunnerContext ctx, Object prompt, String
sessionId, String callId)
+ throws Exception {
+ return new DeferredSubagentFuture(
+ sessionId,
+ callId,
+ ctx,
+ currentTaskRegistry(),
+ () -> prepare(ctx, prompt, sessionId, callId));
+ }
+
+ /**
+ * Prepares one invocation and returns the {@link DurableCallable} running
it: the stable
+ * durable id, the callable running the off-mailbox part, and the optional
recovery reconciler.
+ * Both ids are already assigned; the durable id MUST be derived solely
from the {@code
+ * (sessionId, callId)} pair so it is reproducible after failover.
+ *
+ * <p>Called exactly once per invocation, when the deferred handle is
first resolved, on the
+ * mailbox thread. Implementations may therefore perform the
mailbox-confined part of issuing
+ * the request here (an internal sub-agent sends its call event); the
returned callable's {@link
+ * DurableCallable#call()} carries only the part that runs off the mailbox
thread.
+ *
+ * <p>Implementations that recover an in-flight invocation after failover
supply the reconciler
Review Comment:
nit: following up on the reconciler question from last round. The async base
now wires one by default, and leaving it to the implementor here seems right,
since the deferred mode isn't tied to external services.
Would it be worth spelling out what skipping it costs? Something like:
without a reconciler, a crash between the call landing and its result being
persisted re-invokes on replay. Then it reads as a choice rather than a default.
--
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]
