[api][runtime][python] Add minimal Event lineage

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

@@ -20,14 +20,18 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
+import java.util.function.BiFunction;
 
 /** Base class for all event types in the system. */
 public class Event {
@@ -38,6 +42,9 @@ public class Event {
     private final String type;
     private final Map<String, Object> attributes;
 
+    @Nullable private UUID upstreamEventId;
+    @Nullable private String upstreamActionName;
+
     /**
      * Runtime-internal timestamp from the source record. Not part of the cross-language event
      * contract; used by the Flink runtime for timestamp propagation.
@@ -54,17 +61,38 @@ public Event(String type) {
         this(type, new HashMap<>());
     }
 
+    /**
+     * Reconstructs an Event with an existing identity and optional framework-managed lineage.
+     *
+     * <p>The lineage values support deserialization and reconstruction. When an Action emits this
+     * Event, the runtime overwrites them with the current trigger Event ID and Action name.
+     *
+     * @param id the existing Event ID
+     * @param type the Event type used for routing
+     * @param attributes the Event payload
+     * @param upstreamEventId the ID of the direct upstream Event, or {@code null}
+     * @param upstreamActionName the name of the emitting Action, or {@code null}
+     */
     @JsonCreator
     public Event(
             @JsonProperty("id") UUID id,
             @JsonProperty("type") String type,
-            @JsonProperty("attributes") Map<String, Object> attributes) {
+            @JsonProperty("attributes") Map<String, Object> attributes,
+            @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.upstreamEventId = upstreamEventId;
+        this.upstreamActionName = 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);
     }
 
     public UUID getId() {
@@ -81,6 +109,38 @@ public Map<String, Object> getAttributes() {
         return attributes;
     }
 
+    /** Returns the ID of the Event consumed by the Action that emitted this Event. */
+    @Nullable
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public UUID getUpstreamEventId() {
+        return upstreamEventId;
+    }
+
+    /**
+     * Sets the framework-managed ID of the Event consumed by the emitting Action.
+     *
+     * <p>The runtime overwrites this value when an Action emits the Event.
+     */
+    public void setUpstreamEventId(@Nullable UUID upstreamEventId) {
+        this.upstreamEventId = upstreamEventId;
+    }
+
+    /** Returns the name of the Action that emitted this Event. */
+    @Nullable
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public String getUpstreamActionName() {
+        return upstreamActionName;
+    }
+
+    /**
+     * Sets the framework-managed name of the Action that emitted this Event.
+     *
+     * <p>The runtime overwrites this value when an Action emits the Event.
+     */
+    public void setUpstreamActionName(@Nullable String upstreamActionName) {
+        this.upstreamActionName = upstreamActionName;
+    }
+
     public Object getAttr(String name) {
         return attributes.get(name);
     }
@@ -105,16 +165,38 @@ public void setSourceTimestamp(long timestamp) {
     }
 
     /**
-     * Creates a base Event from another Event, copying id, type, and attributes. Subclasses
-     * override this to reconstruct typed event objects with proper field deserialization.
+     * 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.
      */
     public static Event fromEvent(Event event) {
-        Event copy =
-                new Event(event.getId(), event.getType(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            copy.setSourceTimestamp(event.getSourceTimestamp());
+        return reconstructFrom(
+                event, (id, attributes) -> new Event(id, event.getType(), attributes));
+    }
+
+    /**
+     * Reconstructs a typed Event while preserving the source identity and framework metadata. The
+     * factory receives the source ID and a copy of its attributes.
+     */
+    protected static <T extends Event> T reconstructFrom(
+            Event source, BiFunction<UUID, Map<String, Object>, T> factory) {
+        Objects.requireNonNull(source, "source Event must not be null");
+        Objects.requireNonNull(factory, "Event reconstruction factory must not be null");
+
+        T reconstructed =
+                Objects.requireNonNull(
+                        factory.apply(source.getId(), new HashMap<>(source.getAttributes())),
+                        "Event reconstruction factory must not return null");
+        if (!Objects.equals(source.getId(), reconstructed.getId())) {
+            throw new IllegalStateException(
+                    "Reconstructing the same Event occurrence must preserve Event ID "
+                            + source.getId());
         }
-        return copy;
+        Event reconstructedEvent = reconstructed;
+        reconstructedEvent.sourceTimestamp = source.sourceTimestamp;
+        reconstructedEvent.upstreamEventId = source.upstreamEventId;
+        reconstructedEvent.upstreamActionName = source.upstreamActionName;
+        return reconstructed;
     }
 
     /**

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

@@ -22,7 +22,6 @@
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
 
@@ -50,11 +49,7 @@ public InputEvent(
      * @return a typed InputEvent
      */
     public static InputEvent fromEvent(Event event) {
-        InputEvent result = new InputEvent(event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
-        return result;
+        return reconstructFrom(event, InputEvent::new);
     }
 
     @JsonIgnore

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

@@ -22,7 +22,6 @@
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
 
@@ -53,11 +52,7 @@ public OutputEvent(
      * @return a typed OutputEvent
      */
     public static OutputEvent fromEvent(Event event) {
-        OutputEvent result = new OutputEvent(event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
-        return result;
+        return reconstructFrom(event, OutputEvent::new);
     }
 
     @JsonIgnore

api/src/main/java/org/apache/flink/agents/api/event/ChatRequestEvent.java

@@ -29,7 +29,6 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -96,12 +95,7 @@ private static Map<String, Object> normalizeAttributes(Map<String, Object> attri
      * @return a typed ChatRequestEvent
      */
     public static ChatRequestEvent fromEvent(Event event) {
-        ChatRequestEvent result =
-                new ChatRequestEvent(event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
-        return result;
+        return reconstructFrom(event, ChatRequestEvent::new);
     }
 
     @JsonIgnore

api/src/main/java/org/apache/flink/agents/api/event/ChatResponseEvent.java

@@ -25,7 +25,6 @@
 import org.apache.flink.agents.api.Event;
 import org.apache.flink.agents.api.chat.messages.ChatMessage;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
 
@@ -75,12 +74,7 @@ private static Map<String, Object> normalizeAttributes(Map<String, Object> attri
      * @return a typed ChatResponseEvent
      */
     public static ChatResponseEvent fromEvent(Event event) {
-        ChatResponseEvent result =
-                new ChatResponseEvent(event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
-        return result;
+        return reconstructFrom(event, ChatResponseEvent::new);
     }
 
     @JsonIgnore

api/src/main/java/org/apache/flink/agents/api/event/ContextRetrievalRequestEvent.java

@@ -23,7 +23,6 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.flink.agents.api.Event;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
 
@@ -59,13 +58,7 @@ public ContextRetrievalRequestEvent(
      * @return a typed ContextRetrievalRequestEvent
      */
     public static ContextRetrievalRequestEvent fromEvent(Event event) {
-        ContextRetrievalRequestEvent result =
-                new ContextRetrievalRequestEvent(
-                        event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
-        return result;
+        return reconstructFrom(event, ContextRetrievalRequestEvent::new);
     }
 
     @JsonIgnore

api/src/main/java/org/apache/flink/agents/api/event/ContextRetrievalResponseEvent.java

@@ -26,7 +26,6 @@
 import org.apache.flink.agents.api.vectorstores.Document;
 
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -82,13 +81,7 @@ private static Map<String, Object> normalizeAttributes(Map<String, Object> attri
      * @return a typed ContextRetrievalResponseEvent
      */
     public static ContextRetrievalResponseEvent fromEvent(Event event) {
-        ContextRetrievalResponseEvent result =
-                new ContextRetrievalResponseEvent(
-                        event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
-        return result;
+        return reconstructFrom(event, ContextRetrievalResponseEvent::new);
     }
 
     @JsonIgnore

api/src/main/java/org/apache/flink/agents/api/event/ToolRequestEvent.java

@@ -23,7 +23,6 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.flink.agents.api.Event;
 
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -54,12 +53,7 @@ public ToolRequestEvent(
      */
     @SuppressWarnings("unchecked")
     public static ToolRequestEvent fromEvent(Event event) {
-        ToolRequestEvent result =
-                new ToolRequestEvent(event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
-        return result;
+        return reconstructFrom(event, ToolRequestEvent::new);
     }
 
     @JsonIgnore

api/src/main/java/org/apache/flink/agents/api/event/ToolResponseEvent.java

@@ -98,12 +98,7 @@ private static Map<String, Object> normalizeAttributes(Map<String, Object> attri
      * @return a typed ToolResponseEvent
      */
     public static ToolResponseEvent fromEvent(Event event) {
-        ToolResponseEvent result =
-                new ToolResponseEvent(event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
-        return result;
+        return reconstructFrom(event, ToolResponseEvent::new);
     }
 
     @JsonIgnore