weiqingy commented on code in PR #938:
URL: https://github.com/apache/flink-agents/pull/938#discussion_r3746304855
##########
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:
Option 1 sounds right for this PR. The other two add session machinery while
admitting they don't fix the ordering risk, which was the reason to build them
in the first place.
One thought on where the doc goes. The constraint sits on
`SubagentIdAllocator.nextCallId` (`:143`) today, but what a caller actually
touches is `Subagent.submit(ctx, prompt, sessionId)` (`Subagent.java:37`),
which just says "Issues an invocation under the given sessionId".
`BaseSubagentSetup.submit` pulls the ordinal from the per-task allocator
(`:125`), and `perSessionCallOrdinals` starts fresh each task, so passing the
same session id from a second action gets you `session-1` again.
That matters more in the async mode than it did last round, since the pair
is now the remote key (`submitRequest.getId()` at
`BaseAsyncSubagentSetup.java:126`, and `queryStatus` / `fetchResult` probe the
same pair). Reusing a session id across two actions would have the second call
see the first one's run.
Would putting the "within one action execution" scope on `Subagent.submit`
itself, plus a test, cover it?
--
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]
