weiqingy commented on code in PR #938: URL: https://github.com/apache/flink-agents/pull/938#discussion_r3694826069
########## api/src/main/java/org/apache/flink/agents/api/subagent/BaseSubagentCallable.java: ########## @@ -0,0 +1,65 @@ +/* + * 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.DurableCallable; + +/** + * Convenience base for the {@link DurableCallable} returned by {@code asAsyncCallable}. + * + * <p>Keys the durable call by the framework-assigned identity as {@code sessionId#callId} (the + * {@link SubagentSetup} contract) and captures exceptions thrown by {@link #callInternal()} into + * {@link Result#error(Exception)}, so failures are reported through the result rather than thrown. + * Implementations only provide {@link #callInternal()}. + */ +public abstract class BaseSubagentCallable implements DurableCallable<Result> { + + private final String sessionId; + private final String callId; + + protected BaseSubagentCallable(String sessionId, String callId) { Review Comment: Reading this from the perspective of someone writing an external integration against the new surface. `BaseSubagentCallable` is the convenience base the API steers implementations to (`:23-29`, and `subagent.py:199-201` says so explicitly). It does not override `DurableCallable#reconciler()`, so every sub-agent callable inherits the `null` default at `DurableCallable.java:74`. `durableExecute` selects the reconcile state machine only when `reconciler()` is non-null (`RunnerContextImpl.java:267-275`), and `durableExecuteAsync`, which is the path `SubagentSetup.call` takes, gates identically (`JavaRunnerContextImpl.java:62-70`). Either way sub-agent calls land on `durableExecuteCompletionOnly`. On that path `appendPendingCall` is never reached: its only callers are inside `durableExecuteWithReconcile` (`:555`, `:561`). A crash between "external agent invoked" and "result persisted" therefore leaves no record at all, replay misses the cache, and `call()` re-invokes the external agent. Grepping the new surface, `reconcil` does not appear anywhere under `api/.../subagent/`, in the e2e tests, or in the runtime sub-agent tests. Python surfaces the field (`subagent.py:82`) and forwards it (`:184`), but wires `None` and no test asserts the forwarding. The recovery tests seed only terminal `CallResult`s (`SubagentIdentityRecoveryTest.java:105-108,171-178`), so the pending path has no sub-agent coverage on either side. Concretely: what does an integration author write today to get reconcile-before-resend, and is there anything on this surface that would tell them the option exists? One shape that would make the choice visible, in case it is useful: `BaseSubagentCallable` taking the reconciler as a constructor argument, so passing `null` is something the author decided rather than a default they never saw. ########## python/flink_agents/api/subagent.py: ########## @@ -0,0 +1,202 @@ +################################################################################ +# 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. +################################################################################# +import traceback +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, Callable, Generic, TypeVar + +from flink_agents.api.resource import ResourceType, SerializableResource + +if TYPE_CHECKING: + from flink_agents.api.runner_context import RunnerContext + +T = TypeVar("T") + + +@dataclass +class Result: + """Outcome of a sub-agent call. + + A successful call carries a JSON-serializable ``result``; a failed call + carries a serializable ``error_message`` — the full stack trace of the + failure — rather than a live exception, so a result can be persisted + through durable execution and survive failover. + """ + + success: bool + result: Any = None + error_message: str | None = None + + @staticmethod + def ok(result: Any) -> "Result": + """Create a successful result carrying ``result``.""" + return Result(success=True, result=result) + + @staticmethod + def error(error: BaseException | str) -> "Result": + """Create a failed result from an exception or a plain message. + + For an exception the full stack trace is stored as a serializable + string so the result can survive durable execution. + """ + if isinstance(error, BaseException): + message = "".join( + traceback.format_exception(type(error), error, error.__traceback__) + ) + else: + message = error + return Result(success=False, error_message=message) + + @property + def exception(self) -> Exception | None: + """Reconstruct an exception carrying the stored stack trace; None on success.""" + return None if self.success else RuntimeError(self.error_message) + + +@dataclass +class DurableCallable(Generic[T]): + """A callable for durable execution that carries a stable identifier. + + Used with :meth:`RunnerContext.durable_execute` and + :meth:`RunnerContext.durable_execute_async` so each durable call has a + stable id that persists across job restarts. + """ + + id: str + call: Callable[[], T] + reconciler: Callable[[], T] | None = field(default=None) + + +class BaseSubagentCallable(DurableCallable["Result"], ABC): + """Convenience base for the callable returned by ``as_async_callable``. + + Keys the durable call by the framework-assigned identity as + ``session_id#call_id`` (the :class:`SubagentSetup` contract) and captures + exceptions raised by :meth:`call_internal` into :meth:`Result.error`, so + failures are reported through the result rather than raised. + Implementations only provide :meth:`call_internal`. + """ + + def __init__(self, session_id: str, call_id: str) -> None: + """Initialize with the framework-assigned identity.""" + super().__init__(id=f"{session_id}#{call_id}", call=self._invoke) + + def _invoke(self) -> "Result": + try: + return Result.ok(self.call_internal()) + except Exception as e: + return Result.error(e) + + @abstractmethod + def call_internal(self) -> Any: + """Perform the invocation and return the JSON-serializable payload. + + Raised exceptions are captured into a failed :class:`Result`. + """ + + +class Subagent(ABC): + """Caller-facing interface for a sub-agent invocable from within an action. + + An invocation is identified by a ``(session_id, call_id)`` pair; the + session groups a conversation across invocations. Both ids are assigned by + the framework (:meth:`RunnerContext.next_session_id` and + :meth:`RunnerContext.next_call_id`): callers may supply a session id to + continue a prior session but never supply a call id. + """ + + @abstractmethod + def call( + self, + ctx: "RunnerContext", + prompt: Any, + session_id: str | None = None, + ) -> Result: + """Synchronously invoke the sub-agent and return its :class:`Result`. + + An omitted ``session_id`` starts a new session. + """ + + @abstractmethod + def as_async_callable( + self, + ctx: "RunnerContext", + prompt: Any, + session_id: str | None = None, + ) -> DurableCallable[Result]: + """Return the deferred :class:`DurableCallable` for one invocation. + + An omitted ``session_id`` starts a new session. + """ + + +class SubagentSetup(SerializableResource, Subagent, ABC): + """Base setup for a sub-agent resource, registered as an AGENT resource. + + Hosts the id-resolution chain behind :class:`Subagent`: omitted ids are + assigned via :meth:`RunnerContext.next_session_id` and + :meth:`RunnerContext.next_call_id` before an implementation is ever + invoked, and :meth:`call` runs the deferred callable through durable + execution. Implementations only provide the terminal 4-arg + :meth:`as_async_callable` and contribute nothing to identity assignment. + """ + + @classmethod + def resource_type(cls) -> ResourceType: + """Return resource type of class.""" + return ResourceType.AGENT + + def call( + self, + ctx: "RunnerContext", + prompt: Any, + session_id: str | None = None, + call_id: str | None = None, + ) -> Result: + """Synchronously invoke the sub-agent and return its :class:`Result`. + + Omitted ids are assigned from the context (``call_id`` is + framework-facing and never supplied by callers). The deferred callable + runs through durable execution, so the invocation participates in + failover recovery. + """ + if session_id is None: + session_id = ctx.next_session_id() + if call_id is None: + call_id = ctx.next_call_id(session_id) + callable_ = self.as_async_callable(ctx, prompt, session_id, call_id) + return ctx.durable_execute( Review Comment: `durable_execute` receives `callable_.call`, so the `DurableCallable` and its `id` are dropped at this boundary. `durable_execute` has no id parameter in either the ABC (`runner_context.py:200-207`) or the implementation (`flink_runner_context.py:639-646`), and keys the durable call on `_compute_function_id(func)` plus `_compute_args_digest(args, kwargs)` (`flink_runner_context.py:415-416`). `_invoke` is defined on `BaseSubagentCallable` (`:99`) and bound as `call` in `__init__` (`:97`), so `__qualname__` is identical for every subclass. I loaded this file's class hierarchy and ran the real `_compute_function_id` / `_compute_args_digest` over three callables with different sub-agents, sessions and call ids: ``` ReviewerCall id=sA#c1 function_id=flink_agents.api.subagent.BaseSubagentCallable._invoke args_digest=f744dd20f806e514 ReviewerCall id=sB#c9 function_id=flink_agents.api.subagent.BaseSubagentCallable._invoke args_digest=f744dd20f806e514 CoderCall id=sZ#c3 function_id=flink_agents.api.subagent.BaseSubagentCallable._invoke args_digest=f744dd20f806e514 ``` `args` is always empty on this call path, so the digest is constant as well. Every sub-agent durable call in a Python job collapses to one `(function_id, args_digest)` pair, and only the positional `currentCallIndex` (`RunnerContextImpl.java:789-802`) tells two calls apart. Two consequences, plus one weaker one. The PR's headline property, that a deterministic `(sessionId, callId)` lets failover replay match the right cached result, holds on Java only; Python still matches by call ordinal exactly as before. And `DurableCallable.id` is public surface in Python whose docstring at `:73-77` describes a key nothing reads, its only reader anywhere being the assertion at `test_subagent.py:145`. The weaker one: Java's `matchNextOrClearSubsequentCallResult` mismatch guard cannot fire for Python sub-agent calls, though divergent replay is already documented as undefined behavior at `runner_context.py:218-220`. Was the id meant to reach `durable_execute` here? Adding an `id` / `call_id` parameter to `durable_execute` and `durable_execute_async` and forwarding it as the `function_id` is one way; passing a per-call uniquely-named wrapper instead of the bound method is another. A Python replay test would catch the regression either way, and the harness already exists at `python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py` (fake Java context with `matchNextOrClearSubsequentCallResult` and `recordCallCompletion`). ########## runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java: ########## @@ -582,6 +620,103 @@ protected static class DurableExecutionRuntimeException extends RuntimeException } } + /** + * Caller-side facts identifying one action execution, used as the namespace for deterministic + * sub-agent id assignment: record key, sequence number, caller action name, and the triggering + * event (represented by its type and attributes, so two replays of the same logical event map + * to the same namespace regardless of the event instance id). + */ + public static final class SubagentIdentityNamespace { + + @JsonProperty("key") + private final String key; + + @JsonProperty("sequenceNumber") + private final long sequenceNumber; + + @JsonProperty("actionName") + private final String actionName; + + @JsonProperty("eventType") + private final String eventType; + + @JsonProperty("eventAttributes") + private final Map<String, Object> eventAttributes; + + public SubagentIdentityNamespace( + Object key, long sequenceNumber, String actionName, Event event) { + this.key = key.toString(); + this.sequenceNumber = sequenceNumber; + this.actionName = actionName; + this.eventType = event.getType(); + this.eventAttributes = event.getAttributes(); + } + } + + /** + * Per-{@code ActionTask} context that deterministically assigns sub-agent session and call ids. + * + * <p>The namespace is derived purely from caller-side facts, so a failover replay reproduces + * the same digest and therefore the same id sequence. The context is transient per-task heap + * state: continuation resume carries it forward (ordinals continue), failover rebuilds it + * (ordinals restart). The digest is computed lazily on the first allocation. + */ + public static final class SubagentIdentityContext { + + /** + * Sorts map entries and bean properties so the namespace bytes do not depend on map + * iteration order, which is not guaranteed across JVMs. + */ + private static final ObjectMapper DIGEST_MAPPER = + JsonMapper.builder() + .configure(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS, true) + .configure(MapperFeature.SORT_PROPERTIES_ALPHABETICALLY, true) + .build(); + + private final SubagentIdentityNamespace namespace; + + /** Computed lazily on the first allocation; mailbox-confined, no synchronization. */ + @Nullable private String namespaceDigest; + + private int sessionOrdinal; + private final Map<String, Integer> perSessionCallOrdinals = new HashMap<>(); + + public SubagentIdentityContext( + Object key, long sequenceNumber, String actionName, Event event) { + this.namespace = new SubagentIdentityNamespace(key, sequenceNumber, actionName, event); + } + + /** Creates a new, ordinal-increasing session id scoped to this task's namespace. */ + public String nextSessionId() { + return namespaceDigest() + "-" + (sessionOrdinal++); + } + + /** + * Creates a new call id by appending the per-session ordinal (starting at 1) to the session + * id. Cross-task uniqueness relies on session ids not being shared between action + * executions (see the {@code RunnerContext#nextCallId(String)} contract). + */ + public String nextCallId(String sessionId) { + int ordinal = perSessionCallOrdinals.merge(sessionId, 1, Integer::sum); Review Comment: nit: the javadocs disagree about whether a caller-supplied session id is safe, and both examples take the side this one warns against. `perSessionCallOrdinals` is per-task heap state, so this ordinal restarts at 1 for any session id the task has not seen. The javadoc directly above says cross-task uniqueness "relies on session ids not being shared between action executions (see the `RunnerContext#nextCallId(String)` contract)". The contract it points at states no such thing: `RunnerContext.java:152-153` reads in full, "Creates a new call id for a sub-agent invocation under the given session." `Subagent.java:27-30` goes the other way again, saying callers may supply a session id to continue a prior session. The suite shows the effect without asserting on it. `it1MixedCallsProduceUniqueDeterministicIds` (key `1L`) and `it1RerunningIdenticalInputReproducesIdenticalCaptureSequence` (key `2L`) both produce `explicit-session-checkout-1-1` (`SubagentIdentityIntegrationTest.java:102`), because `ExternalSubagentAgent.java:50` passes `"session-" + prompt` and `external_subagent_agent.py:82` is identical. Nothing breaks today. `CallResult`s live inside an `ActionState` already scoped by `ActionStateUtil.generateKey` (`:44-55`), so two executions never share a list, and neither example passes the id to its endpoint. It would start to matter if an integration used `sessionId#callId` as the remote-side identity, which seems a natural reading of a durable call id. Which javadoc is the contract? If the uniqueness obligation sits on the caller, `Subagent`'s javadoc could say so, and the examples could stop modelling the pattern this one warns against. ########## api/src/main/java/org/apache/flink/agents/api/subagent/Result.java: ########## @@ -0,0 +1,98 @@ +/* + * 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 com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.io.PrintWriter; +import java.io.Serializable; +import java.io.StringWriter; + +/** + * Outcome of a {@link Subagent} call. + * + * <p>Sub-agent implementations should capture internal failures into a {@code Result} (via {@link + * #error}) instead of throwing, so callers can inspect {@link #isSuccess()} without try/catch. + * + * <p>The failure cause is carried as a serializable {@code errorMessage} — the full stack trace of + * the failure — rather than a live exception, so that a {@code Result} can be persisted through + * durable execution. + */ +public class Result implements Serializable { + + private static final long serialVersionUID = 1L; + + private final boolean success; + private final Object result; Review Comment: `result` is typed `Object`, `BaseSubagentCallable.getResultClass()` pins the durable result class to `Result.class` (`BaseSubagentCallable.java:47-48`), and recovery re-binds through the plain `OBJECT_MAPPER` at `RunnerContextImpl.java:516`, which is constructed with no polymorphic typing (`:67-68`). I round-tripped a small POJO payload through this `Result` with `writeValueAsString` / `readValue(s, Result.class)`: ``` serialized: {"success":true,"result":{"verdict":"approve","score":7},"errorMessage":null} payload class after replay: java.util.LinkedHashMap ClassCastException: class java.util.LinkedHashMap cannot be cast to class Review ``` So `getResult()` hands back the author's type on the first execution and a `LinkedHashMap` after a failover replay. The shipped example casts at `ExternalSubagentAgent.java:52` (`((List<?>) result.getResult()).get(0)`) and survives only because JSON arrays bind to `ArrayList`. Every payload in the suite is a `String` or a `List<String>` (`MockExternalSubagentSetup.java:92`, `SubagentIdentityRecoveryTest.java:108`), so nothing currently exercises the shape that breaks. Python does not diverge here. Its durable payload goes through `cloudpickle` (`flink_runner_context.py:430,473`), which preserves the type, so this is also a Java/Python semantic gap on new public API that `AGENTS.md` asks to keep aligned. What should `getResult()` return after a replay when the sub-agent returned a record or a POJO? A couple of routes, in case they help: making `Result` generic and threading the payload class through `getResultClass()`, or keeping the field opaque and adding `getResult(Class<T>)` backed by `OBJECT_MAPPER.convertValue`. Either way a test with a non-`String`, non-collection payload would pin the behavior. ########## api/src/main/java/org/apache/flink/agents/api/subagent/Result.java: ########## @@ -0,0 +1,98 @@ +/* + * 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 com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.io.PrintWriter; +import java.io.Serializable; +import java.io.StringWriter; + +/** + * Outcome of a {@link Subagent} call. + * + * <p>Sub-agent implementations should capture internal failures into a {@code Result} (via {@link + * #error}) instead of throwing, so callers can inspect {@link #isSuccess()} without try/catch. + * + * <p>The failure cause is carried as a serializable {@code errorMessage} — the full stack trace of + * the failure — rather than a live exception, so that a {@code Result} can be persisted through + * durable execution. + */ +public class Result implements Serializable { + + private static final long serialVersionUID = 1L; + + private final boolean success; + private final Object result; + private final String errorMessage; + + @JsonCreator + public Result( + @JsonProperty("success") boolean success, + @JsonProperty("result") Object result, + @JsonProperty("errorMessage") String errorMessage) { + this.success = success; + this.result = result; + this.errorMessage = errorMessage; + } + + /** Creates a successful result carrying the given value. */ + public static Result ok(Object result) { + return new Result(true, result, null); + } + + /** Creates a failed result carrying the full stack trace of the given exception. */ + public static Result error(Exception exception) { + return new Result(false, null, exception == null ? null : stackTraceOf(exception)); Review Comment: `error(Exception)` stores the full stack trace as `errorMessage`, and `BaseSubagentCallable.call()` captures every exception into it rather than throwing (`BaseSubagentCallable.java:52-58`). Two things follow downstream. The durable layer sees a normal completion: `durableExecuteCompletionOnly` calls `recordDurableCompletion` with a null exception (`RunnerContextImpl.java:329`), and `CallResult`'s status is derived as `exceptionPayload == null ? SUCCEEDED : FAILED` (`CallResult.java:100`). So `isFailure()` (`:177-179`) is false for every failed sub-agent call, and `getCurrentCallResultFields()` reports `"SUCCEEDED"` (`RunnerContextImpl.java:490`). Anything keyed on durable-call status reads sub-agent failures as successes. The trace is also uncapped, and `recordCallCompletion` persists the `ActionState` immediately to the configured store (`RunnerContextImpl.java:827-837`, backed by `KafkaActionStateStore` / `FlussActionStateStore`). A flapping external agent writes a multi-KB string per failure into durable storage. Worth capping what gets persisted, say the message plus the top N frames, and keeping the full trace to the log? And is collapsing a captured failure into a SUCCEEDED `CallResult` deliberate, or should the two stay distinguishable at that layer? -- 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]
