[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/ExecutionReporter.java

@@ -62,6 +62,21 @@ void reportExecutionStarted(
             String entityType, String entityName, Map<String, Object> entityMetadata)
             throws Exception;
 
+    /**
+     * 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 previously started logical execution completed successfully.
      *
@@ -72,6 +87,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 +115,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

@@ -56,6 +56,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 +86,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 +124,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() {}

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 >}} |