[Feature][api][runtime] Support auto resolve memory reference for passing data across actions.

api/src/main/java/org/apache/flink/agents/api/Event.java

@@ -22,7 +22,13 @@
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.agents.api.context.MemoryRef;
 
 import javax.annotation.Nullable;
 
@@ -42,6 +48,11 @@ public class Event {
     private final String type;
     private final Map<String, Object> attributes;
 
+    // Keep the annotation on the field as well as the creator parameter so it also applies when
+    // Jackson constructs Event subclasses whose creators do not declare attachments.
+    @JsonDeserialize(contentUsing = AttachmentValueDeserializer.class)
+    private final Map<String, Object> attachments;
+
     @Nullable private UUID upstreamEventId;
     @Nullable private String upstreamActionName;
 
@@ -53,7 +64,7 @@ public class Event {
 
     /** Unified event with user-defined type and attributes. */
     public Event(String type, Map<String, Object> attributes) {
-        this(UUID.randomUUID(), type, attributes);
+        this(UUID.randomUUID(), type, attributes, new HashMap<>());
     }
 
     /** Unified event with user-defined type and empty attributes. */
@@ -78,21 +89,41 @@ public Event(
             @JsonProperty("id") UUID id,
             @JsonProperty("type") String type,
             @JsonProperty("attributes") Map<String, Object> attributes,
+            @JsonProperty("attachments")
+                    @JsonDeserialize(contentUsing = AttachmentValueDeserializer.class)
+                    Map<String, Object> attachments,
             @JsonProperty("upstreamEventId") @Nullable UUID upstreamEventId,
             @JsonProperty("upstreamActionName") @Nullable String upstreamActionName) {
         if (type == null || type.isEmpty()) {
             throw new IllegalArgumentException("Event 'type' must not be null or empty.");
         }
         this.id = id;
         this.type = type;
-        this.attributes = attributes != null ? attributes : new HashMap<>();
+        this.attributes = attributes != null ? new HashMap<>(attributes) : new HashMap<>();
+        this.attachments = attachments != null ? new HashMap<>(attachments) : new HashMap<>();
         this.upstreamEventId = upstreamEventId;
         this.upstreamActionName = upstreamActionName;
     }
 
+    /** Reconstructs an Event with an existing identity, attachments, and no upstream lineage. */
+    public Event(
+            UUID id, String type, Map<String, Object> attributes, Map<String, Object> attachments) {
+        this(id, type, attributes, attachments, null, null);
+    }
+
+    /** Reconstructs an Event with an existing identity and optional framework-managed lineage. */
+    public Event(
+            UUID id,
+            String type,
+            Map<String, Object> attributes,
+            @Nullable UUID upstreamEventId,
+            @Nullable String upstreamActionName) {
+        this(id, type, attributes, new HashMap<>(), upstreamEventId, upstreamActionName);
+    }
+
     /** Reconstructs an Event with an existing identity and no upstream lineage. */
     public Event(UUID id, String type, Map<String, Object> attributes) {
-        this(id, type, attributes, null, null);
+        this(id, type, attributes, new HashMap<>(), null, null);
     }
 
     public UUID getId() {
@@ -109,6 +140,10 @@ public Map<String, Object> getAttributes() {
         return attributes;
     }
 
+    public Map<String, Object> getAttachments() {
+        return attachments;
+    }
+
     /** Returns the ID of the Event consumed by the Action that emitted this Event. */
     @Nullable
     @JsonInclude(JsonInclude.Include.NON_NULL)
@@ -149,6 +184,14 @@ public void setAttr(String name, Object value) {
         attributes.put(name, value);
     }
 
+    public Object getAttachment(String name) {
+        return attachments.get(name);
+    }
+
+    public void setAttachment(String name, Object value) {
+        attachments.put(name, value);
+    }
+
     @JsonIgnore
     public boolean hasSourceTimestamp() {
         return sourceTimestamp != null;
@@ -165,9 +208,9 @@ public void setSourceTimestamp(long timestamp) {
     }
 
     /**
-     * Creates a base Event from another Event, copying its identity, data, and framework metadata.
-     * Subclasses override this to reconstruct typed event objects with proper field
-     * deserialization.
+     * Creates a base Event from another Event, copying its identity, data, attachments, and
+     * framework metadata. Subclasses override this to reconstruct typed event objects with proper
+     * field deserialization.
      */
     public static Event fromEvent(Event event) {
         return reconstructFrom(
@@ -193,6 +236,8 @@ protected static <T extends Event> T reconstructFrom(
                             + source.getId());
         }
         Event reconstructedEvent = reconstructed;
+        reconstructedEvent.attachments.clear();
+        reconstructedEvent.attachments.putAll(source.attachments);
         reconstructedEvent.sourceTimestamp = source.sourceTimestamp;
         reconstructedEvent.upstreamEventId = source.upstreamEventId;
         reconstructedEvent.upstreamActionName = source.upstreamActionName;
@@ -210,18 +255,34 @@ public static Event fromJson(String json) throws IOException {
         return MAPPER.readValue(json, Event.class);
     }
 
+    /** Deserializes one attachment value, preserving explicitly tagged memory references. */
+    static final class AttachmentValueDeserializer extends JsonDeserializer<Object> {
+
+        @Override
+        public Object deserialize(JsonParser parser, DeserializationContext context)
+                throws IOException {
+            JsonNode node = parser.getCodec().readTree(parser);
+            if (node.isObject()
+                    && MemoryRef.TYPE_VALUE.equals(node.path(MemoryRef.TYPE_FIELD).asText())) {
+                return parser.getCodec().treeToValue(node, MemoryRef.class);
+            }
+            return parser.getCodec().treeToValue(node, Object.class);
+        }
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
         Event other = (Event) o;
         return Objects.equals(this.id, other.id)
                 && Objects.equals(this.getType(), other.getType())
-                && Objects.equals(this.attributes, other.attributes);
+                && Objects.equals(this.attributes, other.attributes)
+                && Objects.equals(this.attachments, other.attachments);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(id, getType(), attributes);
+        return Objects.hash(id, getType(), attributes, attachments);
     }
 }

api/src/main/java/org/apache/flink/agents/api/context/MemoryRef.java

@@ -17,17 +17,38 @@
  */
 package org.apache.flink.agents.api.context;
 
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
+import com.fasterxml.jackson.databind.ser.std.StdSerializer;
+
+import java.io.IOException;
 import java.io.Serializable;
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
 import java.util.Objects;
 
 /**
  * A serializable, persistent reference to a specific data item in Short-Term Memory. It acts as a
  * lightweight pointer, containing the path of the data, allowing for efficient passing of large
  * objects between Actions.
  */
+@JsonSerialize(using = MemoryRef.Serializer.class)
+@JsonDeserialize(using = MemoryRef.Deserializer.class)
 public final class MemoryRef implements Serializable {
     private static final long serialVersionUID = 1L;
 
+    public static final String TYPE_FIELD = "@type";
+    public static final String TYPE_VALUE = "memory_ref";
+    public static final String MEMORY_TYPE_FIELD = "memory_type";
+    public static final String PATH_FIELD = "path";
+
     private final MemoryObject.MemoryType type;
     private final String path;
 
@@ -67,6 +88,55 @@ public String getPath() {
         return path;
     }
 
+    public MemoryObject.MemoryType getType() {
+        return type;
+    }
+
+    /** Serializes a {@link MemoryRef} to JSON. */
+    public static final class Serializer extends StdSerializer<MemoryRef> {
+
+        public Serializer() {
+            super(MemoryRef.class);
+        }
+
+        @Override
+        public void serialize(MemoryRef value, JsonGenerator generator, SerializerProvider provider)
+                throws IOException {
+            Map<String, String> serialized = new LinkedHashMap<>();
+            serialized.put(TYPE_FIELD, TYPE_VALUE);
+            serialized.put(MEMORY_TYPE_FIELD, value.getType().name().toLowerCase(Locale.ROOT));
+            serialized.put(PATH_FIELD, value.getPath());
+            generator.writeObject(serialized);
+        }
+    }
+
+    /** Deserializes a {@link MemoryRef} from JSON. */
+    public static final class Deserializer extends StdDeserializer<MemoryRef> {
+
+        public Deserializer() {
+            super(MemoryRef.class);
+        }
+
+        @Override
+        public MemoryRef deserialize(JsonParser parser, DeserializationContext context)
+                throws IOException {
+            JsonNode node = parser.getCodec().readTree(parser);
+            JsonNode typeNode = node.get(MEMORY_TYPE_FIELD);
+            JsonNode pathNode = node.get(PATH_FIELD);
+            if (typeNode == null || typeNode.isNull() || pathNode == null || pathNode.isNull()) {
+                throw new IllegalArgumentException(
+                        "MemoryRef JSON must contain non-null '"
+                                + MEMORY_TYPE_FIELD
+                                + "' and '"
+                                + PATH_FIELD
+                                + "' fields.");
+            }
+            MemoryObject.MemoryType memoryType =
+                    MemoryObject.MemoryType.valueOf(typeNode.asText().toUpperCase(Locale.ROOT));
+            return create(memoryType, pathNode.asText());
+        }
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;