diff --git a/api/src/main/java/org/apache/flink/agents/api/context/RunnerContext.java b/api/src/main/java/org/apache/flink/agents/api/context/RunnerContext.java index a61dc0d18b..76f8b11a1f 100644 --- a/api/src/main/java/org/apache/flink/agents/api/context/RunnerContext.java +++ b/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. * + *

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. * + *

Names used by built-in Action metrics are reserved in this group. + * * @return the individual metric group specific to the current action. */ FlinkAgentsMetricGroup getActionMetricGroup(); diff --git a/api/src/main/java/org/apache/flink/agents/api/trace/ExecutionLifecycleEvents.java b/api/src/main/java/org/apache/flink/agents/api/trace/ExecutionLifecycleEvents.java index dd00a59907..004aa2f595 100644 --- a/api/src/main/java/org/apache/flink/agents/api/trace/ExecutionLifecycleEvents.java +++ b/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); diff --git a/api/src/main/java/org/apache/flink/agents/api/trace/ExecutionReporter.java b/api/src/main/java/org/apache/flink/agents/api/trace/ExecutionReporter.java index a3fb6f11b3..bb049f0aa8 100644 --- a/api/src/main/java/org/apache/flink/agents/api/trace/ExecutionReporter.java +++ b/api/src/main/java/org/apache/flink/agents/api/trace/ExecutionReporter.java @@ -26,7 +26,7 @@ * *

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. + * + *

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 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. + * + *

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 entityMetadata, + String timestamp) + throws Exception { + reportExecutionStarted(entityType, entityName, entityMetadata); + } + + /** + * Reports that a logical execution completed successfully. * *

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 entityMetadata) throws Exception; + /** + * Reports that a logical execution completed successfully at the given occurrence timestamp. + * + *

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 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. + * + *

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 entityMetadata, + Throwable error, + @Nullable String problemCategory, + String timestamp) + throws Exception { + reportExecutionFailed(entityType, entityName, entityMetadata, error, problemCategory); + } } diff --git a/api/src/main/java/org/apache/flink/agents/api/trace/ExecutionReporters.java b/api/src/main/java/org/apache/flink/agents/api/trace/ExecutionReporters.java index 87ca12445e..2a0b3cd021 100644 --- a/api/src/main/java/org/apache/flink/agents/api/trace/ExecutionReporters.java +++ b/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 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 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 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 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) { diff --git a/api/src/main/java/org/apache/flink/agents/api/trace/ToolExecutionMetadataKeys.java b/api/src/main/java/org/apache/flink/agents/api/trace/ToolExecutionMetadataKeys.java index 85f910c9fa..acc6d96c7f 100644 --- a/api/src/main/java/org/apache/flink/agents/api/trace/ToolExecutionMetadataKeys.java +++ b/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() {} diff --git a/api/src/test/java/org/apache/flink/agents/api/trace/ExecutionLifecycleEventsTest.java b/api/src/test/java/org/apache/flink/agents/api/trace/ExecutionLifecycleEventsTest.java index bec417e392..870fec45a6 100644 --- a/api/src/test/java/org/apache/flink/agents/api/trace/ExecutionLifecycleEventsTest.java +++ b/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"); diff --git a/api/src/test/java/org/apache/flink/agents/api/trace/ExecutionReportersTest.java b/api/src/test/java/org/apache/flink/agents/api/trace/ExecutionReportersTest.java index 4699036341..e0dad3a6e5 100644 --- a/api/src/test/java/org/apache/flink/agents/api/trace/ExecutionReportersTest.java +++ b/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( diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md index 4cc35929eb..452ede9e12 100644 --- a/docs/content/docs/operations/configuration.md +++ b/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.

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.
The option value could be: