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

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

@@ -20,9 +20,12 @@
 
 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;
@@ -38,6 +41,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.
@@ -81,6 +87,30 @@ 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 ID of the Event consumed by the Action that emitted this 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 name of the Action that emitted this Event. */
+    public void setUpstreamActionName(@Nullable String upstreamActionName) {
+        this.upstreamActionName = upstreamActionName;
+    }
+
     public Object getAttr(String name) {
         return attributes.get(name);
     }
@@ -105,18 +135,24 @@ 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());
-        }
+        copy.copyFrameworkMetadataFrom(event);
         return copy;
     }
 
+    /** Copies framework-managed metadata when reconstructing a typed Event. */
+    protected void copyFrameworkMetadataFrom(Event event) {
+        this.sourceTimestamp = event.sourceTimestamp;
+        this.upstreamEventId = event.upstreamEventId;
+        this.upstreamActionName = event.upstreamActionName;
+    }
+
     /**
      * Creates an Event from a JSON string.
      *

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

@@ -51,9 +51,7 @@ public InputEvent(
      */
     public static InputEvent fromEvent(Event event) {
         InputEvent result = new InputEvent(event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
+        result.copyFrameworkMetadataFrom(event);
         return result;
     }
 

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

@@ -54,9 +54,7 @@ public OutputEvent(
      */
     public static OutputEvent fromEvent(Event event) {
         OutputEvent result = new OutputEvent(event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
+        result.copyFrameworkMetadataFrom(event);
         return result;
     }
 

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

@@ -98,9 +98,7 @@ private static Map<String, Object> normalizeAttributes(Map<String, Object> attri
     public static ChatRequestEvent fromEvent(Event event) {
         ChatRequestEvent result =
                 new ChatRequestEvent(event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
+        result.copyFrameworkMetadataFrom(event);
         return result;
     }
 

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

@@ -77,9 +77,7 @@ private static Map<String, Object> normalizeAttributes(Map<String, Object> attri
     public static ChatResponseEvent fromEvent(Event event) {
         ChatResponseEvent result =
                 new ChatResponseEvent(event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
+        result.copyFrameworkMetadataFrom(event);
         return result;
     }
 

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

@@ -62,9 +62,7 @@ public static ContextRetrievalRequestEvent fromEvent(Event event) {
         ContextRetrievalRequestEvent result =
                 new ContextRetrievalRequestEvent(
                         event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
+        result.copyFrameworkMetadataFrom(event);
         return result;
     }
 

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

@@ -85,9 +85,7 @@ public static ContextRetrievalResponseEvent fromEvent(Event event) {
         ContextRetrievalResponseEvent result =
                 new ContextRetrievalResponseEvent(
                         event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
+        result.copyFrameworkMetadataFrom(event);
         return result;
     }
 

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

@@ -56,9 +56,7 @@ public ToolRequestEvent(
     public static ToolRequestEvent fromEvent(Event event) {
         ToolRequestEvent result =
                 new ToolRequestEvent(event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
+        result.copyFrameworkMetadataFrom(event);
         return result;
     }
 

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

@@ -100,9 +100,7 @@ private static Map<String, Object> normalizeAttributes(Map<String, Object> attri
     public static ToolResponseEvent fromEvent(Event event) {
         ToolResponseEvent result =
                 new ToolResponseEvent(event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
+        result.copyFrameworkMetadataFrom(event);
         return result;
     }
 

api/src/test/java/org/apache/flink/agents/api/EventTest.java

@@ -143,6 +143,31 @@ void testSubclassedEventJsonRoundTrip() throws Exception {
         assertEquals(InputEvent.EVENT_TYPE, deserialized.getType());
     }
 
+    @Test
+    void testLineageJsonRoundTrip() throws Exception {
+        UUID upstreamEventId = UUID.randomUUID();
+        Event original = new Event("ChildEvent");
+        original.setUpstreamEventId(upstreamEventId);
+        original.setUpstreamActionName("child_action");
+
+        String json = objectMapper.writeValueAsString(original);
+        JsonNode node = objectMapper.readTree(json);
+        Event deserialized = objectMapper.readValue(json, Event.class);
+
+        assertEquals(upstreamEventId.toString(), node.get("upstreamEventId").asText());
+        assertEquals("child_action", node.get("upstreamActionName").asText());
+        assertEquals(upstreamEventId, deserialized.getUpstreamEventId());
+        assertEquals("child_action", deserialized.getUpstreamActionName());
+    }
+
+    @Test
+    void testRootEventOmitsLineageFieldsFromJson() throws Exception {
+        JsonNode node = objectMapper.readTree(objectMapper.writeValueAsString(new InputEvent(1L)));
+
+        assertFalse(node.has("upstreamEventId"));
+        assertFalse(node.has("upstreamActionName"));
+    }
+
     // ── fromJson ───────────────────────────────────────────────────────────
 
     @Test
@@ -218,4 +243,36 @@ void testSourceTimestamp() {
         assertTrue(event.hasSourceTimestamp());
         assertEquals(123456789L, event.getSourceTimestamp());
     }
+
+    @Test
+    void testFromEventCopiesFrameworkMetadata() {
+        UUID upstreamEventId = UUID.randomUUID();
+        Event original = new Event("Test");
+        original.setSourceTimestamp(123456789L);
+        original.setUpstreamEventId(upstreamEventId);
+        original.setUpstreamActionName("test_action");
+
+        Event copy = Event.fromEvent(original);
+
+        assertEquals(123456789L, copy.getSourceTimestamp());
+        assertEquals(upstreamEventId, copy.getUpstreamEventId());
+        assertEquals("test_action", copy.getUpstreamActionName());
+    }
+
+    @Test
+    void testTypedFromEventCopiesLineage() {
+        UUID upstreamEventId = UUID.randomUUID();
+        Event original =
+                new Event(
+                        UUID.randomUUID(),
+                        OutputEvent.EVENT_TYPE,
+                        new HashMap<>(Map.of("output", "result")));
+        original.setUpstreamEventId(upstreamEventId);
+        original.setUpstreamActionName("output_action");
+
+        OutputEvent copy = OutputEvent.fromEvent(original);
+
+        assertEquals(upstreamEventId, copy.getUpstreamEventId());
+        assertEquals("output_action", copy.getUpstreamActionName());
+    }
 }

docs/content/docs/development/workflow_agent.md

@@ -658,11 +658,8 @@ class MyEvent(Event):
     @override
     def from_event(cls, event: Event) -> "MyEvent":
         assert "value" in event.attributes
-        result = MyEvent(value=event.attributes["value"])
-        # Preserve the base event id. Assign it last: the content-based id is
-        # regenerated whenever another field changes.
-        result.id = event.id
-        return result
+        result = cls(value=event.attributes["value"])
+        return result.with_framework_metadata_from(event)
 
     @property
     def value(self) -> str:
@@ -688,12 +685,8 @@ public class MyEvent extends Event {
     }
 
     public static MyEvent fromEvent(Event event) {
-        // Preserve the base event id (and sourceTimestamp) so event logs, listeners,
-        // correlation, deduplication, and timestamp propagation stay consistent.
         MyEvent result = new MyEvent(event.getId(), new HashMap<>(event.getAttributes()));
-        if (event.hasSourceTimestamp()) {
-            result.setSourceTimestamp(event.getSourceTimestamp());
-        }
+        result.copyFrameworkMetadataFrom(event);
         return result;
     }
 
@@ -707,16 +700,15 @@ public class MyEvent extends Event {
 {{< /tabs >}}
 
 {{< hint info >}}
-When reconstructing a typed event, preserve the base `Event` metadata so that event logs,
-listeners, correlation, deduplication, and downstream timestamp propagation stay consistent with
-built-in events:
-
-- **`id`**: copy the source event's `id` onto the reconstructed event, as all built-in events do
-  in both languages. In Python, assign `result.id = event.id` **last**, because the content-based
-  `id` is regenerated whenever any other field changes.
-- **`sourceTimestamp`** (Java only): carry it over with `setSourceTimestamp(...)` when
-  `hasSourceTimestamp()` is true, matching built-in Java events. This field is runtime-internal and
-  used for timestamp propagation; the Python `Event` has no equivalent.
+Typed reconstruction represents the same Event occurrence, so it must preserve the base Event's
+identity and framework-managed metadata:
+
+- **Python**: return `with_framework_metadata_from(event)` after constructing the typed object. It
+  returns a new typed object with the Event's UUIDv4 `id`, `upstream_event_id`, and
+  `upstream_action_name`; the returned Event's `id` remains immutable.
+- **Java**: pass `event.getId()` to the typed constructor, then call
+  `copyFrameworkMetadataFrom(event)`. The helper preserves `sourceTimestamp`,
+  `upstreamEventId`, and `upstreamActionName`.
 {{< /hint >}}
 
 {{< hint info >}}
@@ -756,4 +748,4 @@ private static Map<String, Object> normalizeAttributes(Map<String, Object> attri
 There are several built-in `Event` and `Action` in Flink-Agents:
 * See [Chat Models]({{< ref "docs/development/chat_models#built-in-events-and-actions" >}}) for how to chat with a LLM leveraging built-in action and events.
 * See [Tool Use]({{< ref "docs/development/tool_use#built-in-events-and-actions" >}}) for how to programmatically use a tool leveraging built-in action and events.
-* See [Vector Stores]({{< ref "docs/development/vector_stores#built-in-events-and-actions" >}}) for how to retrieve context from vector stores leveraging built-in action and events.
+* See [Vector Stores]({{< ref "docs/development/vector_stores#built-in-events-and-actions" >}}) for how to retrieve context from vector stores leveraging built-in action and events.

docs/content/docs/operations/monitoring.md

@@ -184,14 +184,44 @@ Each record contains a top-level `timestamp`, the resolved `logLevel`, and a top
 {
   "timestamp": "2024-01-15T10:30:00Z",
   "logLevel": "STANDARD",
-  "eventType": "_input_event",
+  "eventType": "MiddleEvent",
   "event": {
-    "eventType": "_input_event",
-    "...": "..."
+    "eventType": "MiddleEvent",
+    "id": "39361629-4f1d-4b62-b734-a48b181cb6e0",
+    "attributes": {},
+    "type": "MiddleEvent",
+    "upstreamEventId": "dad5ed00-80e2-4746-8bdb-f126bad504b5",
+    "upstreamActionName": "action1"
   }
 }
 ```
 
+`upstreamEventId` identifies the Event consumed by the Action that emitted the current Event, and `upstreamActionName` identifies that Action. The framework maintains both fields directly on the `event` object, outside business `attributes`. A root `InputEvent` omits both fields.
+
+### Trace Tree Reconstruction
+
+The local reader rebuilds InputEvent-rooted Trace Trees from a saved File Event Log. From the repository root, pass either one log file for text output or a log directory for Trace Tree JSON:
+
+```bash
+python tools/reconstruct_trace_tree.py /path/to/events-job-task-0.log
+python tools/reconstruct_trace_tree.py /path/to/event-log-directory --format json
+```
+
+For example, the text output for an `InputEvent -> MiddleEvent -> OutputEvent` lineage is:
+
+```text
+Trace Tree 1
+  _input_event (dad5ed00-80e2-4746-8bdb-f126bad504b5)
+    [Action: action1]
+      MiddleEvent (39361629-4f1d-4b62-b734-a48b181cb6e0)
+        [Action: action2]
+          _output_event (f452ce08-c672-4c9c-841f-d81d15a900c5)
+```
+
+In JSON output, `roots` lists root Event IDs and `nodes` maps each Event ID to its Event node. The reader derives virtual Action nodes from `upstreamActionName`; they are not separate Event Log records. Log order controls display order only and does not add execution-order semantics.
+
+Reconstruction warnings are written to standard error and included in JSON output while valid InputEvent-rooted trees are retained. Warnings cover duplicate Event IDs, missing parent Events, missing Action names, non-InputEvents without parents, and InputEvents with invalid upstream lineage.
+
 ### Event Log Levels
 
 Each event type is logged at a configurable verbosity. Three levels are supported:
@@ -204,7 +234,7 @@ Each event type is logged at a configurable verbosity. Three levels are supporte
 
 The global default is set by [`event-log.level`]({{< ref "docs/operations/configuration#core-options" >}}). At `STANDARD` level, the payload is shrunk along three independent axes — long strings, large arrays, and deep nesting — controlled by `event-log.standard.max-string-length`, `event-log.standard.max-array-elements`, and `event-log.standard.max-depth` respectively. Setting any threshold to `0` disables that specific truncation; setting all three to `0` makes `STANDARD` behave identically to `VERBOSE` (apart from the `logLevel` label). The exact truncation strategy may evolve over time; the contract is only that `STANDARD` keeps logs concise while `VERBOSE` preserves the full payload.
 
-**Fields that are never truncated.** Structural and identifying fields are always preserved in full so log consumers can still group, route, and correlate records: `timestamp`, `logLevel`, top-level `eventType`, and the event's own `id`, `type`, and short scalar fields. Truncation only applies to large nested content (long strings, big arrays, deeply nested objects).
+**Fields that are never truncated.** Structural and identifying fields are always preserved in full so log consumers can still group, route, and correlate records: `timestamp`, `logLevel`, top-level `eventType`, and the event's own `eventType`, `type`, `id`, `upstreamEventId`, and `upstreamActionName`. The `attributes` envelope is also preserved, while large nested content inside it can still be truncated.
 
 **Truncation wrapper format.** When a field is truncated at `STANDARD` level it is replaced by a JSON object that records what was retained and what was dropped. This keeps the record valid JSON and lets downstream tooling detect truncation programmatically:
 

python/flink_agents/api/events/chat_event.py

@@ -83,8 +83,7 @@ def from_event(cls, event: Event) -> "ChatRequestEvent":
             prompt_args=event.attributes.get("prompt_args"),
             output_schema=output_schema_raw,
         )
-        result.id = event.id
-        return result
+        return result.with_framework_metadata_from(event)
 
     @property
     def model(self) -> str:
@@ -160,8 +159,7 @@ def from_event(cls, event: Event) -> "ChatResponseEvent":
             retry_count=event.attributes.get("retry_count", 0),
             total_retry_wait_sec=event.attributes.get("total_retry_wait_sec", 0),
         )
-        result.id = event.id
-        return result
+        return result.with_framework_metadata_from(event)
 
     @property
     def request_id(self) -> UUID:

python/flink_agents/api/events/context_retrieval_event.py

@@ -63,8 +63,7 @@ def from_event(cls, event: Event) -> "ContextRetrievalRequestEvent":
             vector_store=event.attributes["vector_store"],
             max_results=event.attributes.get("max_results", 3),
         )
-        result.id = event.id
-        return result
+        return result.with_framework_metadata_from(event)
 
     @property
     def query(self) -> str:
@@ -124,8 +123,7 @@ def from_event(cls, event: Event) -> "ContextRetrievalResponseEvent":
             query=event.attributes["query"],
             documents=documents,
         )
-        result.id = event.id
-        return result
+        return result.with_framework_metadata_from(event)
 
     @property
     def request_id(self) -> UUID: