[api][runtime][python] Add Agent Trace recording to Event Log

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

@@ -62,7 +62,7 @@ public Event(
         if (type == null || type.isEmpty()) {
             throw new IllegalArgumentException("Event 'type' must not be null or empty.");
         }
-        this.id = id;
+        this.id = id != null ? id : UUID.randomUUID();
         this.type = type;
         this.attributes = attributes != null ? attributes : new HashMap<>();
     }

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

@@ -23,12 +23,12 @@
 
 import java.time.Instant;
 
-/** Contextual information about an event, such as its type and timestamp. */
+/** Runtime context for one observed event occurrence. */
 public class EventContext {
-    /** The routing key for the event, matching the {@code EVENT_TYPE} constant or type string. */
+    /** The event type observed by the runtime, matching {@link Event#getType()}. */
     private final String eventType;
 
-    // Timestamp of when the event occurred
+    /** Timestamp of when the event occurrence was observed by the runtime. */
     private final String timestamp;
 
     public EventContext(Event event) {

api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java

@@ -117,6 +117,14 @@ public class AgentConfigOptions {
     public static final ConfigOption<EventLogLevel> EVENT_LOG_LEVEL =
             new ConfigOption<>("event-log.level", EventLogLevel.class, EventLogLevel.STANDARD);
 
+    /**
+     * Whether to persist Agent Trace context and execution lifecycle Events in the Event Log.
+     * Business Events continue to be logged without trace context when this option is disabled.
+     * In-process execution reporting, including built-in metrics, is unaffected.
+     */
+    public static final ConfigOption<Boolean> EVENT_LOG_TRACE_ENABLED =
+            new ConfigOption<>("event-log.trace.enabled", Boolean.class, false);
+
     /**
      * The maximum string length for event payloads when logging at STANDARD level. Strings
      * exceeding this length will be truncated. Defaults to 2000.

api/src/main/java/org/apache/flink/agents/api/listener/EventListener.java

@@ -22,11 +22,12 @@
 import org.apache.flink.agents.api.EventContext;
 
 /**
- * Interface for event listeners that are notified when events are received for processing.
+ * Interface for event listeners that are notified when business events are received for processing.
  *
  * <p>EventListener provides a callback mechanism triggered at the beginning of event processing.
  * This is useful for monitoring, metrics collection, debugging, or triggering side effects based on
- * event reception.
+ * event reception. Runtime observability records such as execution trace lifecycle events are
+ * written to Event Log but are not delivered to this listener.
  *
  * <p>Event listeners are executed synchronously when an event is received, before any actions are
  * triggered. Implementations should be lightweight and avoid blocking operations to prevent

api/src/main/java/org/apache/flink/agents/api/logger/EventLogger.java

@@ -20,9 +20,12 @@
 
 import org.apache.flink.agents.api.Event;
 import org.apache.flink.agents.api.EventContext;
+import org.apache.flink.agents.api.trace.ExecutionTraceContext;
+
+import javax.annotation.Nullable;
 
 /**
- * Interface for logging action events in the Flink Agents framework.
+ * Interface for logging events in the Flink Agents framework.
  *
  * <p>EventLogger provides a unified interface for capturing, filtering, and persisting events as
  * they flow through the agent execution pipeline. Implementations can target different storage
@@ -49,11 +52,24 @@ public interface EventLogger extends AutoCloseable {
      * should perform the append operation efficiently, as it may be called frequently during event
      * processing.
      *
-     * @param context metadata and other contextual information associated with the event
+     * @param eventContext runtime context associated with the event occurrence
      * @param event the event to be logged
      * @throws Exception if the append operation fails
      */
-    void append(EventContext context, Event event) throws Exception;
+    void append(EventContext eventContext, Event event) throws Exception;
+
+    /**
+     * Appends an event with runtime event context and optional trace context.
+     *
+     * <p>Implementations that persist trace context should override this method. Existing
+     * implementations can keep implementing {@link #append(EventContext, Event)}; the default
+     * behavior preserves the previous contract and ignores the optional trace context.
+     */
+    default void append(
+            EventContext eventContext, Event event, @Nullable ExecutionTraceContext traceContext)
+            throws Exception {
+        append(eventContext, event);
+    }
 
     /**
      * Flush any buffered events to the underlying storage.

api/src/main/java/org/apache/flink/agents/api/tools/ToolExecutionMetadataProvider.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.api.tools;
+
+import java.util.Map;
+
+/**
+ * Optional Tool capability for adding structured metadata to Tool executions.
+ *
+ * <p>The returned metadata is supplemental: it must not define execution identity, parent/child
+ * relationships, or event payload. Implementations should keep values small, stable, and JSON
+ * serializable.
+ */
+public interface ToolExecutionMetadataProvider {
+
+    /**
+     * Returns additional metadata for the current tool call.
+     *
+     * @param parameters tool call parameters associated with the execution
+     * @return small structured metadata to merge into execution {@code entityMetadata}
+     */
+    Map<String, Object> getToolExecutionMetadata(ToolParameters parameters);
+}

api/src/main/java/org/apache/flink/agents/api/trace/ExecutionLifecycleEvents.java

@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.api.trace;
+
+import org.apache.flink.agents.api.Event;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.Map;
+import java.util.Set;
+
+/** Event factory for execution lifecycle reports in the trace model. */
+public final class ExecutionLifecycleEvents {
+
+    public static final String EXECUTION_STARTED_EVENT_TYPE = "_execution_started_event";
+    public static final String EXECUTION_FINISHED_EVENT_TYPE = "_execution_finished_event";
+    public static final String EXECUTION_FAILED_EVENT_TYPE = "_execution_failed_event";
+    public static final String EXECUTION_REUSED_EVENT_TYPE = "_execution_reused_event";
+
+    public static final String STATUS_STARTED = "started";
+    public static final String STATUS_SUCCESS = "success";
+    public static final String STATUS_FAILED = "failed";
+    public static final String STATUS_REUSED = "reused";
+    public static final String STATUS_ATTRIBUTE = "status";
+    public static final String PROBLEM_CATEGORY_ATTRIBUTE = "problemCategory";
+
+    private ExecutionLifecycleEvents() {}
+
+    public static Event executionStarted() {
+        return eventWithStatus(EXECUTION_STARTED_EVENT_TYPE, STATUS_STARTED);
+    }
+
+    public static Event executionFinished() {
+        return eventWithStatus(EXECUTION_FINISHED_EVENT_TYPE, STATUS_SUCCESS);
+    }
+
+    public static Event executionReused() {
+        return eventWithStatus(EXECUTION_REUSED_EVENT_TYPE, STATUS_REUSED);
+    }
+
+    /** Returns whether the given type identifies an execution lifecycle event. */
+    public static boolean isExecutionLifecycleEvent(String eventType) {
+        return EXECUTION_STARTED_EVENT_TYPE.equals(eventType)
+                || EXECUTION_FINISHED_EVENT_TYPE.equals(eventType)
+                || EXECUTION_FAILED_EVENT_TYPE.equals(eventType)
+                || EXECUTION_REUSED_EVENT_TYPE.equals(eventType);
+    }
+
+    public static Event executionFailed(Throwable error) {
+        return executionFailed(error, null);
+    }
+
+    public static Event executionFailed(Throwable error, @Nullable String problemCategory) {
+        Throwable rootCause = rootCause(error);
+        return executionFailed(
+                rootCause.getClass().getName(), rootCause.getMessage(), problemCategory);
+    }
+
+    public static Event executionFailed(
+            String errorType, @Nullable String errorMessage, @Nullable String problemCategory) {
+        Map<String, Object> attributes = new HashMap<>();
+        attributes.put(STATUS_ATTRIBUTE, STATUS_FAILED);
+        if (problemCategory != null) {
+            attributes.put(PROBLEM_CATEGORY_ATTRIBUTE, problemCategory);
+        }
+        if (errorType != null && !errorType.isEmpty()) {
+            attributes.put("errorType", errorType);
+        }
+        if (errorMessage != null && !errorMessage.isEmpty()) {
+            attributes.put("errorMessage", errorMessage);
+        }
+        return new Event(EXECUTION_FAILED_EVENT_TYPE, attributes);
+    }
+
+    private static Event eventWithStatus(String eventType, String status) {
+        Map<String, Object> attributes = new HashMap<>();
+        attributes.put(STATUS_ATTRIBUTE, status);
+        return new Event(eventType, attributes);
+    }
+
+    private static Throwable rootCause(Throwable error) {
+        Throwable current = error;
+        Set<Throwable> visited = Collections.newSetFromMap(new IdentityHashMap<>());
+        while (visited.add(current) && current.getCause() != null) {
+            current = current.getCause();
+        }
+        return current;
+    }
+}

api/src/main/java/org/apache/flink/agents/api/trace/ExecutionReporter.java

@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.api.trace;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
+/**
+ * Optional capability for reporting logical executions nested inside the current action.
+ *
+ * <p>Implementations decide how reports are consumed or ignored. Callers should provide stable
+ * entity type/name pairs and keep metadata small, structured, serializable, and stable for equality
+ * matching between the start and terminal reports of the same logical execution.
+ */
+public interface ExecutionReporter {
+
+    /** Shared entity type names for execution reports. */
+    final class EntityTypes {
+        public static final String ACTION = "action";
+        public static final String LLM = "llm";
+        public static final String PARSER = "parser";
+        public static final String TOOL = "tool";
+
+        private EntityTypes() {}
+    }
+
+    /** Shared low-cardinality problem categories for failed execution reports. */
+    final class ProblemCategories {
+        public static final String ACTION_EXECUTION_FAILED = "action_execution_failed";
+        public static final String MODEL_CALL_FAILED = "model_call_failed";
+        public static final String MODEL_OUTPUT_PARSE_ERROR = "model_output_parse_error";
+        public static final String TOOL_CALL_FAILED = "tool_call_failed";
+
+        private ProblemCategories() {}
+    }
+
+    /**
+     * Reports that a logical execution started within the current action.
+     *
+     * @param entityType stable category of the reported execution, such as LLM, parser, or tool
+     * @param entityName stable name of the reported execution, such as model or tool name
+     * @param entityMetadata small structured metadata used to distinguish repeated executions with
+     *     the same type/name
+     */
+    void reportExecutionStarted(
+            String entityType, String entityName, Map<String, Object> entityMetadata)
+            throws Exception;
+
+    /**
+     * Reports that a previously started logical execution completed successfully.
+     *
+     * <p>The entity type/name/metadata should match the corresponding start report when one was
+     * reported.
+     */
+    void reportExecutionSucceeded(
+            String entityType, String entityName, Map<String, Object> entityMetadata)
+            throws Exception;
+
+    /**
+     * Reports that a logical execution failed.
+     *
+     * <p>The entity type/name/metadata should match the corresponding start report when one was
+     * reported. The problem category should be a stable, low-cardinality classification.
+     */
+    void reportExecutionFailed(
+            String entityType,
+            String entityName,
+            Map<String, Object> entityMetadata,
+            Throwable error,
+            @Nullable String problemCategory)
+            throws Exception;
+}