[runtime][plan][python] Add built-in operational metrics

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

@@ -72,6 +72,8 @@ public interface RunnerContext {
      * from inside a {@link #durableExecute} or {@link #durableExecuteAsync} callable, which runs on
      * a separate thread pool.
      *
+     * <p>Names used by built-in Agent metrics are reserved in this group.
+     *
      * @return the metric group shared across all actions.
      */
     FlinkAgentsMetricGroup getAgentMetricGroup();
@@ -83,6 +85,8 @@ public interface RunnerContext {
      * from inside a {@link #durableExecute} or {@link #durableExecuteAsync} callable, which runs on
      * a separate thread pool.
      *
+     * <p>Names used by built-in Action metrics are reserved in this group.
+     *
      * @return the individual metric group specific to the current action.
      */
     FlinkAgentsMetricGroup getActionMetricGroup();

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

@@ -31,11 +31,13 @@
 /** Event factory for execution lifecycle reports in the trace model. */
 public final class ExecutionLifecycleEvents {
 
+    public static final String EXECUTION_CREATED_EVENT_TYPE = "_execution_created_event";
     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_CREATED = "created";
     public static final String STATUS_STARTED = "started";
     public static final String STATUS_SUCCESS = "success";
     public static final String STATUS_FAILED = "failed";
@@ -45,6 +47,10 @@ public final class ExecutionLifecycleEvents {
 
     private ExecutionLifecycleEvents() {}
 
+    public static Event executionCreated() {
+        return eventWithStatus(EXECUTION_CREATED_EVENT_TYPE, STATUS_CREATED);
+    }
+
     public static Event executionStarted() {
         return eventWithStatus(EXECUTION_STARTED_EVENT_TYPE, STATUS_STARTED);
     }
@@ -59,7 +65,8 @@ public static Event executionReused() {
 
     /** Returns whether the given type identifies an execution lifecycle event. */
     public static boolean isExecutionLifecycleEvent(String eventType) {
-        return EXECUTION_STARTED_EVENT_TYPE.equals(eventType)
+        return EXECUTION_CREATED_EVENT_TYPE.equals(eventType)
+                || EXECUTION_STARTED_EVENT_TYPE.equals(eventType)
                 || EXECUTION_FINISHED_EVENT_TYPE.equals(eventType)
                 || EXECUTION_FAILED_EVENT_TYPE.equals(eventType)
                 || EXECUTION_REUSED_EVENT_TYPE.equals(eventType);

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

@@ -26,7 +26,7 @@
  *
  * <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.
+ * matching between lifecycle reports of the same logical execution.
  */
 public interface ExecutionReporter {
 
@@ -50,6 +50,22 @@ final class ProblemCategories {
         private ProblemCategories() {}
     }
 
+    /**
+     * Reports that a logical execution has been created but has not necessarily started.
+     *
+     * <p>This is an optional lifecycle phase for executions whose admission and invocation are
+     * observably separate. Implementations that do not consume it may keep the default no-op. A
+     * later start or terminal report is not guaranteed, so consumers must not infer whether the
+     * underlying invocation ran from the absence of either report.
+     *
+     * @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 match subsequent lifecycle reports
+     */
+    default void reportExecutionCreated(
+            String entityType, String entityName, Map<String, Object> entityMetadata)
+            throws Exception {}
+
     /**
      * Reports that a logical execution started within the current action.
      *
@@ -63,7 +79,22 @@ void reportExecutionStarted(
             throws Exception;
 
     /**
-     * Reports that a previously started logical execution completed successfully.
+     * Reports that a logical execution started at the given occurrence timestamp.
+     *
+     * <p>The default implementation delegates to {@link #reportExecutionStarted(String, String,
+     * Map)}, so reporters that do not retain occurrence timestamps may use their observation time.
+     */
+    default void reportExecutionStartedAt(
+            String entityType,
+            String entityName,
+            Map<String, Object> entityMetadata,
+            String timestamp)
+            throws Exception {
+        reportExecutionStarted(entityType, entityName, entityMetadata);
+    }
+
+    /**
+     * Reports that a logical execution completed successfully.
      *
      * <p>The entity type/name/metadata should match the corresponding start report when one was
      * reported.
@@ -72,6 +103,21 @@ void reportExecutionSucceeded(
             String entityType, String entityName, Map<String, Object> entityMetadata)
             throws Exception;
 
+    /**
+     * Reports that a logical execution completed successfully at the given occurrence timestamp.
+     *
+     * <p>The default implementation delegates to {@link #reportExecutionSucceeded(String, String,
+     * Map)}, so reporters that do not retain occurrence timestamps may use their observation time.
+     */
+    default void reportExecutionSucceededAt(
+            String entityType,
+            String entityName,
+            Map<String, Object> entityMetadata,
+            String timestamp)
+            throws Exception {
+        reportExecutionSucceeded(entityType, entityName, entityMetadata);
+    }
+
     /**
      * Reports that a logical execution failed.
      *
@@ -85,4 +131,22 @@ void reportExecutionFailed(
             Throwable error,
             @Nullable String problemCategory)
             throws Exception;
+
+    /**
+     * Reports that a logical execution failed at the given occurrence timestamp.
+     *
+     * <p>The default implementation delegates to {@link #reportExecutionFailed(String, String, Map,
+     * Throwable, String)}, so reporters that do not retain occurrence timestamps may use their
+     * observation time.
+     */
+    default void reportExecutionFailedAt(
+            String entityType,
+            String entityName,
+            Map<String, Object> entityMetadata,
+            Throwable error,
+            @Nullable String problemCategory,
+            String timestamp)
+            throws Exception {
+        reportExecutionFailed(entityType, entityName, entityMetadata, error, problemCategory);
+    }
 }

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

@@ -41,6 +41,21 @@ public final class ExecutionReporters {
 
     private ExecutionReporters() {}
 
+    public static void created(RunnerContext ctx, String entityType, String entityName) {
+        created(ctx, entityType, entityName, EMPTY_METADATA);
+    }
+
+    public static void created(
+            RunnerContext ctx,
+            String entityType,
+            String entityName,
+            Map<String, Object> entityMetadata) {
+        report(
+                ctx,
+                reporter -> reporter.reportExecutionCreated(entityType, entityName, entityMetadata),
+                null);
+    }
+
     public static void started(RunnerContext ctx, String entityType, String entityName) {
         started(ctx, entityType, entityName, EMPTY_METADATA);
     }
@@ -56,6 +71,20 @@ public static void started(
                 null);
     }
 
+    public static void startedAt(
+            RunnerContext ctx,
+            String entityType,
+            String entityName,
+            Map<String, Object> entityMetadata,
+            String timestamp) {
+        report(
+                ctx,
+                reporter ->
+                        reporter.reportExecutionStartedAt(
+                                entityType, entityName, entityMetadata, timestamp),
+                null);
+    }
+
     public static void succeeded(RunnerContext ctx, String entityType, String entityName) {
         succeeded(ctx, entityType, entityName, EMPTY_METADATA);
     }
@@ -72,6 +101,20 @@ public static void succeeded(
                 null);
     }
 
+    public static void succeededAt(
+            RunnerContext ctx,
+            String entityType,
+            String entityName,
+            Map<String, Object> entityMetadata,
+            String timestamp) {
+        report(
+                ctx,
+                reporter ->
+                        reporter.reportExecutionSucceededAt(
+                                entityType, entityName, entityMetadata, timestamp),
+                null);
+    }
+
     public static void failed(
             RunnerContext ctx,
             String entityType,
@@ -96,6 +139,27 @@ public static void failed(
                 error);
     }
 
+    public static void failedAt(
+            RunnerContext ctx,
+            String entityType,
+            String entityName,
+            Map<String, Object> entityMetadata,
+            Throwable error,
+            @Nullable String problemCategory,
+            String timestamp) {
+        report(
+                ctx,
+                reporter ->
+                        reporter.reportExecutionFailedAt(
+                                entityType,
+                                entityName,
+                                entityMetadata,
+                                error,
+                                problemCategory,
+                                timestamp),
+                error);
+    }
+
     private static void report(
             RunnerContext ctx, ReporterCall reporterCall, @Nullable Throwable businessError) {
         if (ctx instanceof ExecutionReporter) {

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

@@ -26,6 +26,7 @@ public final class ToolExecutionMetadataKeys {
     public static final String TOOL_TYPE = "toolType";
     public static final String MCP_SERVER = "mcpServer";
     public static final String SKILL_NAME = "skillName";
+    public static final String SKILL_REGISTERED = "skillRegistered";
     public static final String SKILL_RESOURCE_PATH = "skillResourcePath";
 
     private ToolExecutionMetadataKeys() {}

api/src/test/java/org/apache/flink/agents/api/trace/ExecutionLifecycleEventsTest.java

@@ -28,6 +28,17 @@
 /** Tests for {@link ExecutionLifecycleEvents}. */
 class ExecutionLifecycleEventsTest {
 
+    @Test
+    void executionCreatedUsesReservedLifecycleShape() {
+        Event event = ExecutionLifecycleEvents.executionCreated();
+
+        assertThat(event.getType())
+                .isEqualTo(ExecutionLifecycleEvents.EXECUTION_CREATED_EVENT_TYPE);
+        assertThat(event.getAttr(ExecutionLifecycleEvents.STATUS_ATTRIBUTE))
+                .isEqualTo(ExecutionLifecycleEvents.STATUS_CREATED);
+        assertThat(ExecutionLifecycleEvents.isExecutionLifecycleEvent(event.getType())).isTrue();
+    }
+
     @Test
     void executionFailedUsesDeepestCause() {
         IllegalArgumentException root = new IllegalArgumentException("root");

api/src/test/java/org/apache/flink/agents/api/trace/ExecutionReportersTest.java

@@ -36,11 +36,16 @@
 class ExecutionReportersTest {
 
     @Test
-    void startedAndSucceededIgnoreReporterFailures() throws Exception {
+    void createdStartedAndSucceededIgnoreReporterFailures() throws Exception {
         RunnerContext ctx = mockReportingContext();
         ExecutionReporter reporter = (ExecutionReporter) ctx;
+        Exception createdError = new Exception("created failed");
         Exception startedError = new Exception("started failed");
         Exception succeededError = new Exception("succeeded failed");
+        doThrow(createdError)
+                .when(reporter)
+                .reportExecutionCreated(
+                        eq(ExecutionReporter.EntityTypes.LLM), eq("model-a"), anyMap());
         doThrow(startedError)
                 .when(reporter)
                 .reportExecutionStarted(
@@ -50,6 +55,11 @@ void startedAndSucceededIgnoreReporterFailures() throws Exception {
                 .reportExecutionSucceeded(
                         eq(ExecutionReporter.EntityTypes.LLM), eq("model-a"), anyMap());
 
+        assertThatCode(
+                        () ->
+                                ExecutionReporters.created(
+                                        ctx, ExecutionReporter.EntityTypes.LLM, "model-a"))
+                .doesNotThrowAnyException();
         assertThatCode(
                         () ->
                                 ExecutionReporters.started(
@@ -61,6 +71,9 @@ void startedAndSucceededIgnoreReporterFailures() throws Exception {
                                         ctx, ExecutionReporter.EntityTypes.LLM, "model-a"))
                 .doesNotThrowAnyException();
 
+        verify(reporter)
+                .reportExecutionCreated(
+                        eq(ExecutionReporter.EntityTypes.LLM), eq("model-a"), anyMap());
         verify(reporter)
                 .reportExecutionStarted(
                         eq(ExecutionReporter.EntityTypes.LLM), eq("model-a"), anyMap());
@@ -103,6 +116,11 @@ void helpersIgnoreContextsWithoutExecutionReporter() {
         RunnerContext ctx = mock(RunnerContext.class);
         RuntimeException businessError = new RuntimeException("business failed");
 
+        assertThatCode(
+                        () ->
+                                ExecutionReporters.created(
+                                        ctx, ExecutionReporter.EntityTypes.LLM, "model-a"))
+                .doesNotThrowAnyException();
         assertThatCode(
                         () ->
                                 ExecutionReporters.started(

docs/content/docs/operations/configuration.md

@@ -130,7 +130,7 @@ Here is the list of all built-in core configuration options.
 | `action.trigger-condition.evaluate-failure-strategy` | `WARN_AND_SKIP` | ConditionEvaluationFailureStrategy | Handles event-time failures while preparing variables for or evaluating a compiled condition, including a dynamic non-Boolean result. <br/><ul><li>`WARN_AND_SKIP` (default): log a warning, treat that condition as false, and continue with later OR conditions.</li><li>`FAIL`: throw `IllegalStateException` and fail the Flink task; recovery follows the job's restart configuration.</li></ul> Plan-validation failures and runtime compilation or static type-check failures occur during initialization and are not handled by this option. |
 | `error-handling-strategy` | ErrorHandlingStrategy.FAIL | ErrorHandlingStrategy | Strategy for handling errors during model requests, include timeout and unexpected output schema. <br/>The option value could be:<br/> <ul><li>`ErrorHandlingStrategy.FAIL`</li> <li>`ErrorHandlingStrategy.RETRY`</li> <li>`ErrorHandlingStrategy.IGNORE`</li> |
 | `max-retries`             | 3                          | int                   | Number of retries when using `ErrorHandlingStrategy.RETRY`.                                                                                                                                                                                                     |
-| `retry-wait-interval`     | 1                          | int                   | Base wait interval in seconds between retries when using `ErrorHandlingStrategy.RETRY`. Uses exponential backoff: the actual wait time for the Nth retry is `retry-wait-interval * 2^(N-1)` seconds. For example, with default 1s, waits are 1s, 2s, 4s, etc. Retry count and total wait time are reported in `ChatResponseEvent` and recorded as metrics (`retryCount`, `retryWaitSec`) under the connection name. |
+| `retry-wait-interval`     | 1                          | int                   | Base wait interval in seconds between retries when using `ErrorHandlingStrategy.RETRY`. Uses exponential backoff: the actual wait time for the Nth retry is `retry-wait-interval * 2^(N-1)` seconds. For example, with default 1s, waits are 1s, 2s, 4s, etc. Retry count and total wait time are reported in `ChatResponseEvent` and recorded as metrics (`retryCount`, `retryWaitSec`) under the configured ChatModel resource name. |
 | `chat.async`              | true                       | boolean               | Whether chat asynchronously for built-in chat action.                                                                                                                                                                                                           |
 | `tool-call.async`         | true                       | boolean               | Whether the built-in tool-call action runs each tool via durable async execution.                                                                                                                                                                               |
 | `tool-call.parallelism`   | os cpu count               | int                   | In-flight concurrency for tool calls from one `ToolRequestEvent` batch when `tool-call.async` is enabled. `1` runs tools serially; values greater than `1` run a parallel durable batch with a sliding window of at most that many concurrent tool calls. On **Java**, concurrent in-batch execution requires **JDK 21+** (Continuation API); below JDK 21 the batch still runs but tool calls execute serially. **Python** uses the shared async `ThreadPoolExecutor` and runs batches concurrently regardless of JDK version. Increases in-flight external calls; after failover, unfinished tools may be submitted again — side-effecting tools should be idempotent or provide a reconciler. {{< hint warning >}}**Default is parallel** (`os cpu count`). Chat, RAG, and tool batches share one `num-async-threads` pool **per operator subtask** (all keys on that subtask). Built-in actions for a single key run one at a time, so chat and a tool batch on the **same key** do not overlap in the usual chat → tool path; delay shows up mainly **across keys** on the same subtask. With defaults (`num-async-threads = 2× cores`, `tool-call.parallelism = cores`), one batch can use up to half the pool; several busy keys can still saturate it. Lower this value or increase `num-async-threads` on hot subtasks. {{< /hint >}} |