pltbkd commented on code in PR #938:
URL: https://github.com/apache/flink-agents/pull/938#discussion_r3702008392
##########
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:
**Contract question — direct answer:** The RunnerContextImpl warning that
"session ids [should] not be shared between action executions" is the correct
contract. The current identity context is scoped to a single action task —
Subagent.java's "callers may supply a session id to continue a prior session"
should be qualified as "within the same action execution," and the
RunnerContext interface's `nextCallId` should also document this scope.
Root cause: when session id / call id counting is reused across action
tasks, branch/diamond structures may cause the ordering at first acquisition to
differ from the ordering at replay, making sessionId replayability
unguaranteeable.
---
Taking this nit as a starting point, I'd like to discuss what infrastructure
external subagent should provide for cross-action session management. Currently
`nextSessionId` / `nextCallId` are framework built-in tools, and SubagentSetup
can already be overridden to use a custom id management mechanism. However, the
ordering-induced replayability problem persists in every approach. Here are the
directions I see:
**Option 1: Doc + tests only**
Codify the contract as: "Anonymous Session can by default only continue
conversation within the current Action; cross-Action / cross-record
conversation continuation requires the Subagent implementation to provide that
capability itself." The framework provides no cross-action infrastructure;
implementations decide how to manage sessions.
**Option 2: Unified override in BaseExternalSubagent**
Override `nextSessionId` / `nextCallId` in BaseExternalSubagent, maintaining
its own sessionId → session-info mapping instead of using the context's
built-in tools. This gives external subagents out-of-the-box cross-actionTask
session management. The ordering risk remains.
**Option 3: getSession interface**
Replace the single `submit` with `getSession(sessionId<optional>)`, then
`submit` on the same session instance to continue the conversation. The session
instance internally maintains the ordinal; the subagent manages session
instances and eviction timing itself, and the same session can be retrieved
from different tasks. This is more natural, but the actual implementation is
similar to Option 2 — it introduces a Session object for session-level state,
adds complexity for single-call subagents, and still cannot resolve the
cross-Action ordering risk.
How far do you think we should go?
--
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]
