pltbkd commented on code in PR #938: URL: https://github.com/apache/flink-agents/pull/938#discussion_r3765898165
########## 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: I'll make it 500 by default. This value affects not only the probe frequency but also the latency bound. Since it's sensitive on both ends, would it be worth introducing an option to encourage users to overwrite it for their own service? -- 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]
