pltbkd commented on code in PR #1220: URL: https://github.com/apache/flink-agents/pull/1220#discussion_r4278742203
########## runtime/src/main/java/org/apache/flink/agents/runtime/subgraph/SubgraphEvent.java: ########## @@ -0,0 +1,80 @@ +/* + * 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.subgraph; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonProperty; +import org.apache.flink.agents.api.Event; + +import java.util.Map; +import java.util.UUID; + +/** + * Private event envelope shared by all internal graph calls. Only bootstraps can be dispatched + * while their producing action is suspended. The delegate is what a business action receives. + * + * <p>The invocation and delegate identities participate in action-state lookup. A bootstrap's + * delegate must have a replay-stable id; graph outputs retain their ids in completed ActionState. + * This does not resolve the caller-identity collision tracked by issue 1084. + */ +public final class SubgraphEvent extends Event { + public static final String EVENT_TYPE = "_internal_subgraph"; + private static final String INVOCATION_ID = "invocation_id"; + private static final String BOOTSTRAP = "bootstrap"; + private static final String DELEGATE_TYPE = "delegate_type"; + private static final String DELEGATE_ATTRIBUTES = "delegate_attributes"; + private static final String DELEGATE_ID = "delegate_id"; + private final Event delegate; + + public SubgraphEvent(String invocationId, Event delegate, boolean bootstrap) { + super(EVENT_TYPE); + this.delegate = delegate; + setAttr(INVOCATION_ID, invocationId); + setAttr(BOOTSTRAP, bootstrap); + setAttr(DELEGATE_TYPE, delegate.getType()); + setAttr(DELEGATE_ATTRIBUTES, delegate.getAttributes()); + setAttr(DELEGATE_ID, delegate.getId().toString()); Review Comment: I suggest leaving the deterministic id strategy for SubgraphEvent and the delegated events unchanged, since we plan to revisit it in #1084. So we don't need to set DELEGATE_ID here, nor compute a deterministic eventId for the delegated event. ########## runtime/src/main/java/org/apache/flink/agents/runtime/chat/ChatCallOwner.java: ########## @@ -0,0 +1,67 @@ +/* + * 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.chat; + +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.runtime.subagent.SubagentIdAllocator; + +/** + * Caller key and chat-call counter for one action execution: one action processing one event. This + * is not an action definition, an executor, or the state of an individual chat invocation. Multiple + * chat calls issued during that execution share this owner, but each has its own {@link + * ChatInvocation}. + * + * <p>The action task's context creates and retains this object. Suspending and resuming the action, + * including transferring its context to a generated successor task, preserves the same owner. + * Switching the shared runner context to another action installs that action's owner instead. + * + * <p>{@link #nextCallId()} assigns deterministic IDs from the caller's key, sequence number, action + * name, event, and call ordinal. On failure recovery this owner is recreated; replaying the same + * caller facts and call order reproduces the same call IDs for durable-result lookup. The owner + * itself is not checkpointed. The caller key also lets the manager clean up abandoned calls when + * record processing finishes. + * + * <p>Call ID allocation is confined to the action/mailbox thread. + */ +public final class ChatCallOwner { + private final SubagentIdAllocator ids; Review Comment: It's a bit strange to have a SubagentIdAllocator here. Maybe we could simply make it a common component by renaming it to DeterministicIdAllocator, ReplayableIdAllocator, or something similar? ########## runtime/src/main/java/org/apache/flink/agents/runtime/subagent/InternalSubagentCallStatus.java: ########## @@ -15,157 +15,63 @@ * 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.Event; +import org.apache.flink.agents.api.OutputEvent; +import org.apache.flink.agents.plan.actions.Action; import org.apache.flink.agents.runtime.context.RunnerContextImpl; +import org.apache.flink.agents.runtime.subgraph.SubgraphDefinition; +import org.apache.flink.agents.runtime.subgraph.SubgraphInvocation; -import javax.annotation.Nullable; - -import java.util.ArrayList; import java.util.List; -import java.util.concurrent.CompletableFuture; -import java.util.function.Supplier; +import java.util.stream.Collectors; /** - * Self-coordinating quiesce state machine for a single sub-agent call. Completion is driven by two - * counters: {@code runningActions} (child actions executing) and {@code pendingEvents} (child - * events not yet dispatched). Every mutating method checks whether both reached zero and completes - * the response future itself, so callers never invoke a separate tryComplete step. + * Subagent-specific result adapter. The common invocation owns action accounting and completion; + * the setup retains this reference only to implement session/call lookup for public handles. */ -public class InternalSubagentCallStatus { - - private final String callId; - private final String scope; - private final String sessionId; - private final InternalSubagentSetup setup; - private final CompletableFuture<List<Object>> responseFuture = new CompletableFuture<>(); - private int runningActions; - private int pendingEvents; - private final List<Object> output = new ArrayList<>(); - - /** - * Isolated memory view shared by every child action of this call. The first child action - * creates it and the rest reuse it, so a value one child action writes stays readable by the - * next action of the same call. It lives exactly as long as this call status: the owning setup - * drops both when the record finishes. - */ - @Nullable private RunnerContextImpl.MemoryContext isolatedMemoryContext; - +public final class InternalSubagentCallStatus extends SubgraphInvocation<List<Object>> { Review Comment: Better to rename it to InternalSubagentInvocation? -- 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]
