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 a61dc0d18..76f8b11a1 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/ExecutionReporter.java b/api/src/main/java/org/apache/flink/agents/api/trace/ExecutionReporter.java
index a3fb6f11b..95e75ad6d 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
@@ -62,6 +62,21 @@ void reportExecutionStarted(
String entityType, String entityName, Map entityMetadata)
throws Exception;
+ /**
+ * 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 previously started logical execution completed successfully.
*
@@ -72,6 +87,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 +115,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 87ca12445..4b5a12984 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
@@ -56,6 +56,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 +86,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 +124,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 85f910c9f..acc6d96c7 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/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md
index 4feaf250b..f23e03ac8 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.
`WARN_AND_SKIP` (default): log a warning, treat that condition as false, and continue with later OR conditions.
`FAIL`: throw `IllegalStateException` and fail the Flink task; recovery follows the job's restart configuration.
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:
`ErrorHandlingStrategy.FAIL`
`ErrorHandlingStrategy.RETRY`
`ErrorHandlingStrategy.IGNORE`
|
| `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 >}} |
diff --git a/docs/content/docs/operations/monitoring.md b/docs/content/docs/operations/monitoring.md
index 652c9bfe9..bcd72eae1 100644
--- a/docs/content/docs/operations/monitoring.md
+++ b/docs/content/docs/operations/monitoring.md
@@ -26,7 +26,7 @@ under the License.
### Built-in Metrics
-We offer data monitoring for built-in metrics, which includes events, actions, and token usage.
+We offer data monitoring for built-in metrics, including input runs, events, actions, execution health, and token usage.
#### Event and Action Metrics
@@ -36,11 +36,64 @@ We offer data monitoring for built-in metrics, which includes events, actions, a
| **Agent** | numOfEventProcessedPerSec | The number of Events this operator has processed per second. | Meter |
| **Agent** | numOfActionsExecuted | The total number of actions this operator has executed. | Count |
| **Agent** | numOfActionsExecutedPerSec | The number of actions this operator has executed per second. | Meter |
+| **Agent** | numOfInputRunsSucceeded | The number of input runs that reached the run-completion boundary. | Count |
+| **Agent** | numOfInputRunsFailed | The number of input runs terminated by an unhandled exception. | Count |
+| **Agent** | inputRunLatencyMs | End-to-end input-run latency from entering the agent operator to completion or failure, including time queued behind another input with the same key. | Histogram |
+| **Agent** | inputRunQueueLatencyMs | Time from entering the agent operator until the input run starts processing. | Histogram |
+| **Agent** | inputRunProcessingLatencyMs | Time from the input-run start boundary until completion or failure. | Histogram |
+| **Agent** | numOfPendingInputEvents | Current number of input Events buffered behind an active run with the same key. | Gauge |
+| **Agent** | numOfActiveInputRuns | Current number of logical input runs that are processing or waiting for asynchronous work. | Gauge |
| **Action** | action.\.numOfActionsExecuted | The total number of actions this operator has executed for a specific action name. | Count |
| **Action** | action.\.numOfActionsExecutedPerSec | The number of actions this operator has executed per second for a specific action name. | Meter |
+| **Action** | action.\.actionSchedulingLatencyMs | Time from enqueuing the initial Action task until it is selected for execution. | Histogram |
+| **Action** | action.\.actionExecutionLatencyMs | End-to-end latency of one logical Action execution, including asynchronous waits and continuations. | Histogram |
+| **Action** | action.\.numOfPendingActionTasks | Current number of physical Action task segments waiting to run, including continuations. | Gauge |
+| **Action** | action.\.numOfActiveActionExecutions | Current number of logical Action executions that have started but have not reached a terminal state. | Gauge |
| **Agent** | eventLogTruncatedEvents | Number of event log records whose payload was truncated at `STANDARD` level. Increments once per event, regardless of how many fields inside it were truncated. Use this to decide whether to raise truncation thresholds or move specific event types to `VERBOSE`. | Count |
| **Agent** | eventLogWriteFailures | Number of Event Log write attempts for which `append`, `flush`, or both failed. Event Log writes are best-effort and do not fail the job. | Count |
+For a locally observed input run, `inputRunLatencyMs` is split into queueing and processing time at the input-run start boundary. `numOfPendingInputEvents` counts buffered inputs, while `numOfActiveInputRuns` counts logical runs; an asynchronous run remains active while it is waiting for its continuation.
+
+An Action execution can be active while one of its continuation tasks is pending, so `numOfActiveActionExecutions` and `numOfPendingActionTasks` are independent. Action scheduling latency is recorded only for the initial task; continuation queueing does not create another scheduling sample.
+
+Input-run outcomes and all latency samples are process-local. Runs or Action executions already in flight when a task is restored do not produce latency samples because their original timestamps are unavailable. An input Event restored from the pending queue can still produce an outcome and processing-latency sample after it starts in the new task attempt, but it does not produce queue or end-to-end latency. Current-count gauges are rebuilt from Flink state after restore.
+
+#### Execution Metrics
+
+LLM and Tool outcome and latency metrics are derived from execution lifecycle Events. Each Tool callable records its own start and completion timestamps; its durable execution Outcome determines the reported result, including failures during result persistence. Events may be delivered after the parallel batch completes, but use each call's timestamps rather than the batch duration. Event publication is independent of response aggregation, so a later response-processing failure does not repeat or discard reports for calls with available Outcomes. The `model_resource`, `tool`, `skill`, and `mcp_server` scopes are independent key-value scopes directly under an Action; none is nested under another. The existing `model` scope remains dedicated to model usage metrics.
+
+| Scope | Metrics | Description | Type |
+|-------|---------|-------------|------|
+| **Model Resource** | action.\.model_resource.\.numOfLlmCallsSucceeded | The number of framework-observed model invocations that returned successfully. | Count |
+| **Model Resource** | action.\.model_resource.\.numOfLlmCallsFailed | The number of framework-observed model invocations that failed. | Count |
+| **Model Resource** | action.\.model_resource.\.llmCallLatencyMs | Latency of each framework-observed model invocation, excluding structured-output parsing and retry wait time. | Histogram |
+| **Model Resource** | action.\.model_resource.\.retryCount | The number of additional model invocations initiated when `ErrorHandlingStrategy.RETRY` is configured. Only recorded when at least one retry occurs. See [retry-wait-interval]({{< ref "docs/operations/configuration#core-options" >}}). | Count |
+| **Model Resource** | action.\.model_resource.\.retryWaitSec | The total backoff time, in seconds, accumulated when `ErrorHandlingStrategy.RETRY` is configured. Only recorded when at least one retry occurs. | Count |
+| **Tool** | action.\.tool.\.numOfToolCallsSucceeded | The number of successful calls to the Tool. | Count |
+| **Tool** | action.\.tool.\.numOfToolCallsFailed | The number of failed calls to the Tool. | Count |
+| **Tool** | action.\.tool.\.toolCallLatencyMs | Time spent invoking the individual Tool, excluding time waiting for other calls in the same parallel batch. | Histogram |
+| **Skill** | action.\.skill.\.numOfSkillLoads | The number of terminal explicit `load_skill` calls attributed to the Skill, regardless of outcome. | Count |
+| **Skill** | action.\.skill.\.skillLoadLatencyMs | Time spent invoking an explicit `load_skill` call. | Histogram |
+| **MCP Server** | action.\.mcp_server.\.numOfMcpToolCallsSucceeded | The number of successful Tool calls served by the MCP Server. | Count |
+| **MCP Server** | action.\.mcp_server.\.numOfMcpToolCallsFailed | The number of failed Tool calls served by the MCP Server. | Count |
+| **MCP Server** | action.\.mcp_server.\.mcpToolCallLatencyMs | Individual Tool invocation latency aggregated across the MCP Server. | Histogram |
+
+An LLM metric represents one framework invocation of `ChatModel`. A framework retry that calls the model again produces another LLM outcome and latency sample; retries hidden inside a provider or connection are not observed. Every named Tool execution emits Tool metrics. Skill metrics are emitted only for explicit `load_skill` calls; subsequent Tool calls are not inferred to belong to a Skill. MCP metrics aggregate only Tool executions carrying an explicit MCP Server resource name. A `load_skill` or MCP Tool execution therefore contributes to both its Tool scope and the corresponding Skill or MCP Server scope.
+
+Execution metrics currently inherit Agent Trace's durable-replay behavior. During fine-grained recovery, a cached durable LLM or Tool result is reported as a new successful execution because child cache reuse is not exposed to execution reporting. The corresponding success counter therefore increments. A cached LLM result may produce a near-zero latency sample; a cached Tool result produces no latency sample because the Tool callable was not invoked and no execution duration was measured. Distinguishing reused child executions is follow-up work.
+
+Tool names that are not registered runtime resources are aggregated under the fixed `tool=unknown` scope. Requested Skill names that do not resolve in the runtime registry are similarly aggregated under `skill=unknown`. The original requested names remain available in Agent Trace records, while Metric scope cardinality remains bounded.
+
+Tool outcomes follow the existing language-specific Tool contracts. In both Java and Python, resource preparation, invocation, or durable result-persistence exceptions are failures. Java additionally treats an unsuccessful `ToolResponse` as a failed Tool execution. Python Tools return arbitrary values and currently have no equivalent explicit error-result type, so the runtime does not infer failure from a normally returned Python value when durable execution also succeeds.
+
+`numOfSkillLoads` counts terminal calls rather than successful loads. Under the current Tool contracts, a `load_skill` not-found response returns normally and is therefore observed as a successful Tool outcome. Consequently, Tool and MCP outcome metrics use the same names and scopes in both runtimes, but explicit error-result semantics are not yet identical. This alignment, including explicit failure results for framework Tools such as `load_skill`, is tracked in [Issue #956](https://github.com/apache/flink-agents/issues/956).
+
+Execution latency tracking is process-local and uses the occurrence timestamps in matching start and terminal Events. Both Events must be observed in the same task attempt. A Tool latency sample additionally requires the Tool callable to start no later than the Action observes its durable result; queueing and parallel-batch fan-in are excluded. LLM and Tool terminal counters are still updated when no matching start Event is available.
+
+A request timeout does not necessarily stop a running Tool. ToolCallAction records when the durable call returns or raises, before processing responses or publishing Events. If a Tool has not finished by that time, its terminal Event uses this fixed observation time; a later Tool completion cannot extend it or produce another terminal Event. This excludes response-processing and Event-publication delays, but is not the execution framework's exact timeout-decision time: any delay before the durable result reaches the Action remains included. A callable that starts after this observation retains its timeout terminal Event without a start Event or latency sample, even if it later completes in the background. Cached results and failures before invocation also retain terminal-only reporting. If a batch aborts without returning per-call Outcomes, known starts may be reported without terminal Events; no terminal execution Events are inferred from timestamps or the batch exception. Existing business ToolResponseEvent error handling is unchanged. These observation rules do not change the execution framework's timeout, failure, or durable-persistence behavior.
+
+In previous releases, `retryCount` and `retryWaitSec` used the `model.` scope. They now use `model_resource.` so retries are attributed to the configured ChatModel resource. Existing queries and dashboards for these two metrics must use the new scope.
+
#### Token Usage Metrics
Token usage metrics are automatically recorded when chat models are invoked through `ChatModelConnection`. These metrics help track LLM API usage and costs.
@@ -49,13 +102,13 @@ Token usage metrics are automatically recorded when chat models are invoked thro
|-----------|--------------------------------------------------------------|--------------------------------------------------------------------------------|-------|
| **Model** | action.\.model.\.promptTokens | The total number of prompt tokens consumed by the model within an action. | Count |
| **Model** | action.\.model.\.completionTokens | The total number of completion tokens generated by the model within an action. | Count |
-| **Model** | action.\.model.\.retryCount | The total number of retries performed for model requests when using `ErrorHandlingStrategy.RETRY`. Only recorded when at least one retry occurs. See [retry-wait-interval]({{< ref "docs/operations/configuration#core-options" >}}). | Count |
-| **Model** | action.\.model.\.retryWaitSec | The total wait time (in seconds) spent across retries for model requests when using `ErrorHandlingStrategy.RETRY`. | Count |
### How to add custom metrics
In Flink Agents, users implement their logic by defining custom Actions that respond to various Events throughout the Agent lifecycle. To support user-defined metrics, we introduce two new properties: `agent_metric_group` and `action_metric_group` in the RunnerContext. These properties allow users to create or update global metrics and independent metrics for actions. For an introduction to metric types, please refer to the [Metric types documentation](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/ops/metrics/#metric-types).
+Metric names listed in the built-in tables above are reserved in their corresponding scopes. Custom metrics must use different names within the same scope.
+
Here is the user case example:
{{< tabs "Custom Metrics" >}}
@@ -116,7 +169,7 @@ public class MyAgent extends Agent {
### How to check the metrics with Flink executor
-Flink agents enable the reporting of metrics to external systems by creating a metric identifier prefix in the format `.taskmanager....`. Agent-specific metrics use key-value metric groups (e.g., `action.`, `model.`) which are exposed as dimensions/labels in reporters that support them (such as Prometheus). Please refer to [Flink Metric Reporters](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/deployment/metric_reporters/) for more details.
+Flink agents enable the reporting of metrics to external systems by creating a metric identifier prefix in the format `.taskmanager....`. For an agent operator, `` is the agent name. If the Agent name is unavailable, the operator retains the previous `action-execute-operator` value as a fallback. This changes only the value of the existing `` scope; the Agent-specific metric hierarchy is unchanged. Queries and dashboards that filter on `operator_name=action-execute-operator` must use the Agent name after upgrading. Agent-specific metrics use key-value metric groups (e.g., `action.`, `model.`) which are exposed as dimensions/labels in reporters that support them (such as Prometheus). Please refer to [Flink Metric Reporters](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/deployment/metric_reporters/) for more details.
Additionally, we can check the metric results in the Flink Job WebUI using the metric identifier prefix `.`.
diff --git a/e2e-test/flink-agents-end-to-end-tests-integration/src/test/java/org/apache/flink/agents/integration/test/TokenMetricsE2ETest.java b/e2e-test/flink-agents-end-to-end-tests-integration/src/test/java/org/apache/flink/agents/integration/test/TokenMetricsE2ETest.java
index 5b0982b29..7e4006e41 100644
--- a/e2e-test/flink-agents-end-to-end-tests-integration/src/test/java/org/apache/flink/agents/integration/test/TokenMetricsE2ETest.java
+++ b/e2e-test/flink-agents-end-to-end-tests-integration/src/test/java/org/apache/flink/agents/integration/test/TokenMetricsE2ETest.java
@@ -69,12 +69,16 @@ class TokenMetricsE2ETest {
+ "\"finish_reason\":\"stop\"}],"
+ "\"usage\":{\"prompt_tokens\":10,\"completion_tokens\":5,\"total_tokens\":15}}";
+ private static final String OPERATOR_NAME = TokenMetricsE2EAgent.class.getSimpleName();
+
private static final Pattern PREFIX_PATTERN =
Pattern.compile(
- "^\\.taskmanager\\.([a-f0-9-]+)\\.Flink Streaming Job\\.action-execute-operator\\.0\\.");
+ "^\\.taskmanager\\.([a-f0-9-]+)\\.Flink Streaming Job\\."
+ + Pattern.quote(OPERATOR_NAME)
+ + "\\.0\\.");
private static final String PREFIX_TEMPLATE =
- ".taskmanager.%s.Flink Streaming Job.action-execute-operator.0.";
+ ".taskmanager.%s.Flink Streaming Job." + OPERATOR_NAME + ".0.";
/**
* Expected agent Counter metrics. Each key is the deterministic suffix after the prefix in the
diff --git a/plan/src/main/java/org/apache/flink/agents/plan/actions/ChatModelAction.java b/plan/src/main/java/org/apache/flink/agents/plan/actions/ChatModelAction.java
index 57c191c75..cd4ec6edf 100644
--- a/plan/src/main/java/org/apache/flink/agents/plan/actions/ChatModelAction.java
+++ b/plan/src/main/java/org/apache/flink/agents/plan/actions/ChatModelAction.java
@@ -203,15 +203,16 @@ private static Map getRetryStats(MemoryObject sensoryMem, UUID ini
}
private static void recordRetryMetrics(
- RunnerContext ctx, String model, int retryCount, int totalRetryWaitSec) {
+ RunnerContext ctx, String modelResource, int retryCount, int totalRetryWaitSec) {
if (retryCount <= 0) {
return;
}
FlinkAgentsMetricGroup metricGroup = ctx.getActionMetricGroup();
if (metricGroup != null) {
- FlinkAgentsMetricGroup modelGroup = metricGroup.getSubGroup("model", model);
- modelGroup.getCounter("retryCount").inc(retryCount);
- modelGroup.getCounter("retryWaitSec").inc(totalRetryWaitSec);
+ FlinkAgentsMetricGroup modelResourceGroup =
+ metricGroup.getSubGroup("model_resource", modelResource);
+ modelResourceGroup.getCounter("retryCount").inc(retryCount);
+ modelResourceGroup.getCounter("retryWaitSec").inc(totalRetryWaitSec);
}
}
@@ -412,7 +413,7 @@ private static void chat(
recordAttemptRetryStats(
ctx,
initialRequestId,
- result.chatModel,
+ result.model,
result.retryCount,
result.totalRetryWaitSec);
if (selection.isRouter) {
@@ -481,7 +482,7 @@ private static void chat(
return;
} catch (ChatModelInvoker.ChatAttemptFailed e) {
recordAttemptRetryStats(
- ctx, initialRequestId, e.chatModel, e.retryCount, e.totalRetryWaitSec);
+ ctx, initialRequestId, e.model, e.retryCount, e.totalRetryWaitSec);
// Keep every candidate's failure: chain the previous error into the new one so
// exhaustion surfaces A's and B's errors as suppressed of C's, not just C's.
if (lastError != null && lastError != e.error) {
@@ -525,7 +526,7 @@ private static void chat(
private static void recordAttemptRetryStats(
RunnerContext ctx,
UUID initialRequestId,
- BaseChatModelSetup chatModel,
+ String modelResource,
int retryCount,
int retryWaitSec)
throws Exception {
@@ -533,12 +534,7 @@ private static void recordAttemptRetryStats(
return;
}
accumulateRetryStats(ctx.getSensoryMemory(), initialRequestId, retryCount, retryWaitSec);
- String metricModel = chatModel == null ? null : chatModel.getConnectionName();
- recordRetryMetrics(
- ctx,
- metricModel == null || metricModel.isEmpty() ? "unknown" : metricModel,
- retryCount,
- retryWaitSec);
+ recordRetryMetrics(ctx, modelResource, retryCount, retryWaitSec);
}
/**
diff --git a/plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java b/plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java
index a67571a8f..dc6779319 100644
--- a/plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java
+++ b/plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java
@@ -42,6 +42,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.Instant;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
@@ -133,9 +134,6 @@ private static List buildToolCallExecutions(
name,
tool,
metadataParameters);
- ExecutionReporters.started(
- ctx, ExecutionReporter.EntityTypes.TOOL, name, entityMetadata);
-
if (tool == null || preparationError != null) {
Exception failure =
preparationError != null
@@ -169,6 +167,7 @@ private static List buildToolCallExecutions(
final Tool toolRef = tool;
final Map callArguments = mergedArguments;
+ ToolCallOccurrence occurrence = new ToolCallOccurrence();
DurableCallable callable =
new DurableCallable<>() {
@Override
@@ -183,10 +182,15 @@ public Class getResultClass() {
@Override
public ToolResponse call() throws Exception {
- return toolRef.call(new ToolParameters(callArguments));
+ occurrence.markStarted();
+ try {
+ return toolRef.call(new ToolParameters(callArguments));
+ } finally {
+ occurrence.markFinished();
+ }
}
};
- executions.add(new ToolCallExecution(id, name, callable, entityMetadata));
+ executions.add(new ToolCallExecution(id, name, callable, entityMetadata, occurrence));
}
return executions;
}
@@ -201,26 +205,29 @@ private static void executeParallel(
for (ToolCallExecution execution : executions) {
callables.add(execution.callable);
}
+ List> outcomes = List.of();
+ Instant resultObservedAt = null;
try {
- List> outcomes = ctx.durableExecuteAllAsync(callables);
+ outcomes = ctx.durableExecuteAllAsync(callables);
+ resultObservedAt = Instant.now();
for (int i = 0; i < outcomes.size(); i++) {
- recordOutcome(executions.get(i), outcomes.get(i), ctx, success, error, responses);
+ recordOutcome(executions.get(i), outcomes.get(i), success, error, responses);
}
} catch (Exception e) {
+ if (resultObservedAt == null) {
+ resultObservedAt = Instant.now();
+ }
for (ToolCallExecution execution : executions) {
recordExecutionException(execution, e, success, error, responses);
}
- } catch (Error e) {
- for (ToolCallExecution execution : executions) {
- ExecutionReporters.failed(
+ } finally {
+ for (int i = 0; i < executions.size(); i++) {
+ reportExecution(
+ executions.get(i),
ctx,
- ExecutionReporter.EntityTypes.TOOL,
- execution.name,
- execution.entityMetadata,
- e,
- ExecutionReporter.ProblemCategories.TOOL_CALL_FAILED);
+ i < outcomes.size() ? outcomes.get(i) : null,
+ resultObservedAt);
}
- throw e;
}
}
@@ -232,45 +239,24 @@ private static void executeSequentially(
Map error,
Map responses) {
for (ToolCallExecution execution : executions) {
+ Outcome outcome = null;
+ Instant resultObservedAt = null;
try {
ToolResponse response =
toolCallAsync
? ctx.durableExecuteAsync(execution.callable)
: ctx.durableExecute(execution.callable);
+ resultObservedAt = Instant.now();
+ outcome = Outcome.success(response);
recordToolResponse(execution.id, response, success, error, responses);
- if (response.isError()) {
- ExecutionReporters.failed(
- ctx,
- ExecutionReporter.EntityTypes.TOOL,
- execution.name,
- execution.entityMetadata,
- new RuntimeException(response.getError()),
- ExecutionReporter.ProblemCategories.TOOL_CALL_FAILED);
- } else {
- ExecutionReporters.succeeded(
- ctx,
- ExecutionReporter.EntityTypes.TOOL,
- execution.name,
- execution.entityMetadata);
- }
} catch (Exception e) {
+ if (resultObservedAt == null) {
+ resultObservedAt = Instant.now();
+ }
+ outcome = Outcome.failure(e);
recordExecutionException(execution, e, success, error, responses);
- ExecutionReporters.failed(
- ctx,
- ExecutionReporter.EntityTypes.TOOL,
- execution.name,
- execution.entityMetadata,
- e,
- ExecutionReporter.ProblemCategories.TOOL_CALL_FAILED);
- } catch (Error e) {
- ExecutionReporters.failed(
- ctx,
- ExecutionReporter.EntityTypes.TOOL,
- execution.name,
- execution.entityMetadata,
- e,
- ExecutionReporter.ProblemCategories.TOOL_CALL_FAILED);
- throw e;
+ } finally {
+ reportExecution(execution, ctx, outcome, resultObservedAt);
}
}
}
@@ -278,38 +264,65 @@ private static void executeSequentially(
private static void recordOutcome(
ToolCallExecution execution,
Outcome outcome,
- RunnerContext ctx,
Map success,
Map error,
Map responses) {
if (outcome.isFailure()) {
recordExecutionException(execution, outcome.getError(), success, error, responses);
- ExecutionReporters.failed(
+ } else {
+ recordToolResponse(execution.id, outcome.getValue(), success, error, responses);
+ }
+ }
+
+ private static void reportExecution(
+ ToolCallExecution execution,
+ RunnerContext ctx,
+ Outcome outcome,
+ Instant resultObservedAt) {
+ Instant finishedAt = execution.occurrence.finishedAt;
+ Instant startedAt = execution.occurrence.startedAt;
+ if (startedAt != null && (outcome == null || !startedAt.isAfter(resultObservedAt))) {
+ ExecutionReporters.startedAt(
+ ctx,
+ ExecutionReporter.EntityTypes.TOOL,
+ execution.name,
+ execution.entityMetadata,
+ startedAt.toString());
+ }
+ if (outcome == null) {
+ return;
+ }
+ // A timed-out callable may finish after the Action already received its failure.
+ if (finishedAt == null || finishedAt.isAfter(resultObservedAt)) {
+ finishedAt = resultObservedAt;
+ }
+ Throwable failure =
+ outcome.isFailure() ? outcome.getError() : toolResponseFailure(outcome.getValue());
+
+ if (failure == null) {
+ ExecutionReporters.succeededAt(
ctx,
ExecutionReporter.EntityTypes.TOOL,
execution.name,
execution.entityMetadata,
- outcome.getError(),
- ExecutionReporter.ProblemCategories.TOOL_CALL_FAILED);
+ finishedAt.toString());
} else {
- ToolResponse response = outcome.getValue();
- recordToolResponse(execution.id, response, success, error, responses);
- if (response.isError()) {
- ExecutionReporters.failed(
- ctx,
- ExecutionReporter.EntityTypes.TOOL,
- execution.name,
- execution.entityMetadata,
- new RuntimeException(response.getError()),
- ExecutionReporter.ProblemCategories.TOOL_CALL_FAILED);
- } else {
- ExecutionReporters.succeeded(
- ctx,
- ExecutionReporter.EntityTypes.TOOL,
- execution.name,
- execution.entityMetadata);
- }
+ ExecutionReporters.failedAt(
+ ctx,
+ ExecutionReporter.EntityTypes.TOOL,
+ execution.name,
+ execution.entityMetadata,
+ failure,
+ ExecutionReporter.ProblemCategories.TOOL_CALL_FAILED,
+ finishedAt.toString());
+ }
+ }
+
+ private static Throwable toolResponseFailure(ToolResponse response) {
+ if (response == null) {
+ return new IllegalStateException("Tool returned a null response.");
}
+ return response.isError() ? new RuntimeException(response.getError()) : null;
}
private static void recordInlineResponse(
@@ -356,16 +369,32 @@ private static final class ToolCallExecution {
private final String name;
private final DurableCallable callable;
private final Map entityMetadata;
+ private final ToolCallOccurrence occurrence;
private ToolCallExecution(
String id,
String name,
DurableCallable callable,
- Map entityMetadata) {
+ Map entityMetadata,
+ ToolCallOccurrence occurrence) {
this.id = id;
this.name = name;
this.callable = callable;
this.entityMetadata = entityMetadata;
+ this.occurrence = occurrence;
+ }
+ }
+
+ private static final class ToolCallOccurrence {
+ private volatile Instant startedAt;
+ private volatile Instant finishedAt;
+
+ private void markStarted() {
+ startedAt = Instant.now();
+ }
+
+ private void markFinished() {
+ finishedAt = Instant.now();
}
}
diff --git a/plan/src/test/java/org/apache/flink/agents/plan/actions/ChatModelActionRetryTest.java b/plan/src/test/java/org/apache/flink/agents/plan/actions/ChatModelActionRetryTest.java
index a1d7b0cf3..c9ee0513c 100644
--- a/plan/src/test/java/org/apache/flink/agents/plan/actions/ChatModelActionRetryTest.java
+++ b/plan/src/test/java/org/apache/flink/agents/plan/actions/ChatModelActionRetryTest.java
@@ -323,8 +323,8 @@ void chatRetriesWithExponentialBackoff() throws Exception {
assertThat(responseEvent.getTotalRetryWaitSec()).isEqualTo(1);
assertThat(elapsed).isGreaterThanOrEqualTo(1000L);
- // Verify metrics recorded under connection name
- verify(mockActionMetricGroup).getSubGroup("model", mockChatModel.getConnectionName());
+ // Retry health belongs to the ChatModel resource, not the provider connection or model.
+ verify(mockActionMetricGroup).getSubGroup("model_resource", "test-model");
verify(mockRetryCountCounter).inc(1);
verify(mockRetryWaitSecCounter).inc(1);
}
@@ -351,6 +351,9 @@ void chatExhaustsRetriesAndThrows() {
.hasMessage("persistent error");
assertThat(sentEvents).isEmpty();
+ verify(mockActionMetricGroup).getSubGroup("model_resource", "test-model");
+ verify(mockRetryCountCounter).inc(2);
+ verify(mockRetryWaitSecCounter).inc(0);
}
@Test
diff --git a/plan/src/test/java/org/apache/flink/agents/plan/actions/ChatModelActionRoutingTest.java b/plan/src/test/java/org/apache/flink/agents/plan/actions/ChatModelActionRoutingTest.java
index 5964a2051..25bcb17a7 100644
--- a/plan/src/test/java/org/apache/flink/agents/plan/actions/ChatModelActionRoutingTest.java
+++ b/plan/src/test/java/org/apache/flink/agents/plan/actions/ChatModelActionRoutingTest.java
@@ -46,6 +46,8 @@
import org.apache.flink.agents.api.resource.ResourceType;
import org.apache.flink.agents.api.tools.ToolResponse;
import org.apache.flink.agents.plan.AgentConfiguration;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Histogram;
import org.junit.jupiter.api.Test;
import java.util.ArrayDeque;
@@ -60,6 +62,9 @@
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
/** Integration tests for model routing inside {@link ChatModelAction}. */
public class ChatModelActionRoutingTest {
@@ -117,6 +122,7 @@ static class FakeRunnerContext implements RunnerContext {
private final ModelRouter router;
private final MemoryObject sensoryMemory = new FakeMemoryObject(new HashMap<>());
private final AgentConfiguration config = new AgentConfiguration(Map.of());
+ private FlinkAgentsMetricGroup actionMetricGroup;
FakeRunnerContext(ModelRouter router) {
this.router = router;
@@ -144,6 +150,11 @@ FakeRunnerContext withRetryBudget(int maxRetries, int waitIntervalSec) {
return this;
}
+ FakeRunnerContext withActionMetricGroup(FlinkAgentsMetricGroup actionMetricGroup) {
+ this.actionMetricGroup = actionMetricGroup;
+ return this;
+ }
+
@Override
public boolean hasResource(String name, ResourceType type) {
return type == ResourceType.MODEL_ROUTER && "router".equals(name) && router != null;
@@ -191,7 +202,7 @@ public FlinkAgentsMetricGroup getAgentMetricGroup() {
@Override
public FlinkAgentsMetricGroup getActionMetricGroup() {
- return null;
+ return actionMetricGroup;
}
@Override
@@ -394,6 +405,17 @@ void routedRequestUsesRoutedDurableCallIds() throws Exception {
@Test
void retryBudgetRunsBeforeFallback() throws Exception {
+ FlinkAgentsMetricGroup actionMetricGroup = mock(FlinkAgentsMetricGroup.class);
+ FlinkAgentsMetricGroup modelResourceMetricGroup = mock(FlinkAgentsMetricGroup.class);
+ Counter retryCount = mock(Counter.class);
+ Counter retryWaitSec = mock(Counter.class);
+ when(actionMetricGroup.getHistogram("routingDecisionLatencyMs"))
+ .thenReturn(mock(Histogram.class));
+ when(actionMetricGroup.getSubGroup("model_resource", "big"))
+ .thenReturn(modelResourceMetricGroup);
+ when(modelResourceMetricGroup.getCounter("retryCount")).thenReturn(retryCount);
+ when(modelResourceMetricGroup.getCounter("retryWaitSec")).thenReturn(retryWaitSec);
+
ModelRouter router =
new ModelRouter(
ModelRouter.of("small", "big")
@@ -406,6 +428,7 @@ void retryBudgetRunsBeforeFallback() throws Exception {
new FakeRunnerContext(router)
.withErrorHandling(Agent.ErrorHandlingStrategy.RETRY)
.withRetryBudget(1, 0)
+ .withActionMetricGroup(actionMetricGroup)
.register(
"big",
new FakeChatModel(
@@ -422,6 +445,9 @@ void retryBudgetRunsBeforeFallback() throws Exception {
assertThat(ctx.chatResponse().getResponse().getContent()).isEqualTo("recovered on retry");
assertThat(ctx.resolvedChatModels).containsExactly("big");
assertThat(ctx.routingEventCount()).isEqualTo(1L);
+ verify(actionMetricGroup).getSubGroup("model_resource", "big");
+ verify(retryCount).inc(1);
+ verify(retryWaitSec).inc(0);
}
@Test
diff --git a/plan/src/test/java/org/apache/flink/agents/plan/actions/ToolCallActionReportTest.java b/plan/src/test/java/org/apache/flink/agents/plan/actions/ToolCallActionReportTest.java
index 08348327b..08ff3f088 100644
--- a/plan/src/test/java/org/apache/flink/agents/plan/actions/ToolCallActionReportTest.java
+++ b/plan/src/test/java/org/apache/flink/agents/plan/actions/ToolCallActionReportTest.java
@@ -20,6 +20,7 @@
import org.apache.flink.agents.api.Event;
import org.apache.flink.agents.api.agents.AgentExecutionOptions;
import org.apache.flink.agents.api.context.DurableCallable;
+import org.apache.flink.agents.api.context.Outcome;
import org.apache.flink.agents.api.context.RunnerContext;
import org.apache.flink.agents.api.event.ToolRequestEvent;
import org.apache.flink.agents.api.event.ToolResponseEvent;
@@ -33,18 +34,34 @@
import org.apache.flink.agents.api.trace.ExecutionReporter;
import org.apache.flink.agents.api.trace.ToolExecutionMetadataKeys;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.mockito.Mockito.withSettings;
@@ -86,10 +103,22 @@ void processToolRequestReportsEachToolCall() throws Exception {
metadata.put(ToolExecutionMetadataKeys.TOOL_TYPE, ToolType.MCP.getValue());
metadata.put(ToolExecutionMetadataKeys.MCP_SERVER, "search-server");
ExecutionReporter reporter = (ExecutionReporter) ctx;
+ ArgumentCaptor startedAt = ArgumentCaptor.forClass(String.class);
+ ArgumentCaptor finishedAt = ArgumentCaptor.forClass(String.class);
verify(reporter)
- .reportExecutionStarted(ExecutionReporter.EntityTypes.TOOL, "search", metadata);
+ .reportExecutionStartedAt(
+ eq(ExecutionReporter.EntityTypes.TOOL),
+ eq("search"),
+ eq(metadata),
+ startedAt.capture());
verify(reporter)
- .reportExecutionSucceeded(ExecutionReporter.EntityTypes.TOOL, "search", metadata);
+ .reportExecutionSucceededAt(
+ eq(ExecutionReporter.EntityTypes.TOOL),
+ eq("search"),
+ eq(metadata),
+ finishedAt.capture());
+ assertThat(Instant.parse(finishedAt.getValue()))
+ .isAfterOrEqualTo(Instant.parse(startedAt.getValue()));
assertThat(sentEvents).hasSize(1);
assertThat(sentEvents.get(0)).isInstanceOf(ToolResponseEvent.class);
@@ -123,20 +152,482 @@ void processToolRequestMarksErrorResponseAsFailed() throws Exception {
metadata.put(ToolExecutionMetadataKeys.TOOL_CALL_ID, "call-1");
ExecutionReporter reporter = (ExecutionReporter) ctx;
verify(reporter)
- .reportExecutionFailed(
+ .reportExecutionFailedAt(
eq(ExecutionReporter.EntityTypes.TOOL),
eq("search"),
eq(metadata),
any(Throwable.class),
- eq(ExecutionReporter.ProblemCategories.TOOL_CALL_FAILED));
+ eq(ExecutionReporter.ProblemCategories.TOOL_CALL_FAILED),
+ anyString());
verify(reporter, never())
- .reportExecutionSucceeded(ExecutionReporter.EntityTypes.TOOL, "search", metadata);
+ .reportExecutionSucceededAt(anyString(), anyString(), anyMap(), anyString());
ToolResponseEvent responseEvent = (ToolResponseEvent) sentEvents.get(0);
assertThat(responseEvent.getSuccess()).containsEntry("call-1", false);
assertThat(responseEvent.getError()).containsEntry("call-1", "tool rejected request");
}
+ @Test
+ void parallelToolCallsReportIndependentOutcomes() throws Exception {
+ RunnerContext ctx =
+ mock(RunnerContext.class, withSettings().extraInterfaces(ExecutionReporter.class));
+ List sentEvents = new ArrayList<>();
+ Tool tool = mock(Tool.class);
+ when(tool.call(any()))
+ .thenAnswer(
+ invocation -> {
+ String query =
+ invocation
+ .getArgument(0)
+ .getParameter("query", String.class);
+ if ("call-2".equals(query)) {
+ throw new IllegalStateException("call-2 failed");
+ }
+ if ("call-3".equals(query)) {
+ return ToolResponse.error("call-3 rejected");
+ }
+ return ToolResponse.success("ok");
+ });
+ when(ctx.getResource("search", ResourceType.TOOL)).thenReturn(tool);
+ when(ctx.getConfig()).thenReturn(toolCallConfig(true, 3));
+ when(ctx.durableExecuteAllAsync(any()))
+ .thenAnswer(
+ invocation -> {
+ List> callables =
+ invocation.getArgument(0);
+ List> outcomes = new ArrayList<>();
+ for (DurableCallable callable : callables) {
+ try {
+ outcomes.add(Outcome.success(callable.call()));
+ } catch (Exception e) {
+ outcomes.add(Outcome.failure(e));
+ }
+ }
+ return outcomes;
+ });
+ doAnswer(inv -> sentEvents.add(inv.getArgument(0))).when(ctx).sendEvent(any());
+
+ ToolCallAction.processToolRequest(
+ new ToolRequestEvent(
+ "test-model",
+ List.of(toolCall("call-1"), toolCall("call-2"), toolCall("call-3"))),
+ ctx);
+
+ ExecutionReporter reporter = (ExecutionReporter) ctx;
+ verify(reporter, times(3))
+ .reportExecutionStartedAt(
+ eq(ExecutionReporter.EntityTypes.TOOL),
+ eq("search"),
+ anyMap(),
+ anyString());
+ verify(reporter)
+ .reportExecutionSucceededAt(
+ eq(ExecutionReporter.EntityTypes.TOOL),
+ eq("search"),
+ anyMap(),
+ anyString());
+ verify(reporter, times(2))
+ .reportExecutionFailedAt(
+ eq(ExecutionReporter.EntityTypes.TOOL),
+ eq("search"),
+ anyMap(),
+ any(Throwable.class),
+ eq(ExecutionReporter.ProblemCategories.TOOL_CALL_FAILED),
+ anyString());
+
+ ToolResponseEvent response = (ToolResponseEvent) sentEvents.get(0);
+ assertThat(response.getSuccess())
+ .containsEntry("call-1", true)
+ .containsEntry("call-2", false)
+ .containsEntry("call-3", false);
+ assertThat(response.getError())
+ .containsEntry("call-2", "call-2 failed")
+ .containsEntry("call-3", "call-3 rejected");
+ }
+
+ @Test
+ void responseProcessingFailureDoesNotRepeatCompletedOccurrences() throws Exception {
+ Tool tool = mock(Tool.class);
+ when(tool.call(any()))
+ .thenAnswer(
+ invocation ->
+ "call-2"
+ .equals(
+ invocation
+ .getArgument(0)
+ .getParameter(
+ "query", String.class))
+ ? null
+ : ToolResponse.success("ok"));
+ RunnerContext ctx = parallelContext(tool);
+
+ ToolCallAction.processToolRequest(parallelRequest(), ctx);
+
+ assertReports(
+ ctx,
+ List.of("call-1", "call-2", "call-3"),
+ List.of("call-1", "call-3"),
+ List.of("call-2"));
+ assertBusinessResponsesFailed(ctx);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void durableFailureIsReportedAsToolFailure(boolean async) throws Exception {
+ IllegalStateException failure = new IllegalStateException("persist failed");
+ Tool tool = mock(Tool.class);
+ when(tool.call(any())).thenReturn(ToolResponse.success("ok"));
+ RunnerContext ctx = parallelContext(tool);
+ when(ctx.getConfig()).thenReturn(toolCallConfig(async, 1));
+ when(ctx.durableExecute(any()))
+ .thenAnswer(
+ invocation -> {
+ invocation.>getArgument(0).call();
+ throw failure;
+ });
+ when(ctx.durableExecuteAsync(any()))
+ .thenAnswer(
+ invocation -> {
+ invocation.>getArgument(0).call();
+ throw failure;
+ });
+
+ ToolCallAction.processToolRequest(parallelRequest(), ctx);
+
+ assertReports(
+ ctx,
+ List.of("call-1", "call-2", "call-3"),
+ List.of(),
+ List.of("call-1", "call-2", "call-3"));
+ verify((ExecutionReporter) ctx, times(3))
+ .reportExecutionFailedAt(
+ anyString(), anyString(), anyMap(), eq(failure), anyString(), anyString());
+ assertBusinessResponsesFailed(ctx);
+ }
+
+ @Test
+ void parallelDurableFailureIsReportedForItsToolCall() throws Exception {
+ IllegalStateException failure = new IllegalStateException("persist failed");
+ Tool tool = mock(Tool.class);
+ when(tool.call(any())).thenReturn(ToolResponse.success("ok"));
+ RunnerContext ctx = parallelContext(tool);
+ doAnswer(
+ invocation -> {
+ List> callables =
+ invocation.getArgument(0);
+ List> outcomes = new ArrayList<>();
+ for (DurableCallable callable : callables) {
+ outcomes.add(Outcome.success(callable.call()));
+ }
+ outcomes.set(1, Outcome.failure(failure));
+ return outcomes;
+ })
+ .when(ctx)
+ .durableExecuteAllAsync(any());
+
+ ToolCallAction.processToolRequest(parallelRequest(), ctx);
+
+ assertReports(
+ ctx,
+ List.of("call-1", "call-2", "call-3"),
+ List.of("call-1", "call-3"),
+ List.of("call-2"));
+ verify((ExecutionReporter) ctx)
+ .reportExecutionFailedAt(
+ anyString(), anyString(), anyMap(), eq(failure), anyString(), anyString());
+ ArgumentCaptor event = ArgumentCaptor.forClass(Event.class);
+ verify(ctx).sendEvent(event.capture());
+ assertThat(((ToolResponseEvent) event.getValue()).getSuccess())
+ .containsEntry("call-1", true)
+ .containsEntry("call-2", false)
+ .containsEntry("call-3", true);
+ }
+
+ @Test
+ void timeoutIsReportedAsFailureWithoutRepeatingOnLateCompletion() throws Exception {
+ TimeoutException failure = new TimeoutException("request timed out");
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ ExecutorService worker = Executors.newSingleThreadExecutor();
+ AtomicReference> pending = new AtomicReference<>();
+ AtomicReference reportingStartedAt = new AtomicReference<>();
+ Tool tool = mock(Tool.class);
+ when(tool.call(any()))
+ .thenAnswer(
+ invocation -> {
+ started.countDown();
+ assertThat(release.await(5, TimeUnit.SECONDS)).isTrue();
+ return ToolResponse.success("ok");
+ });
+ RunnerContext ctx = parallelContext(tool);
+ when(ctx.getConfig()).thenReturn(toolCallConfig(true, 1));
+ when(ctx.durableExecuteAsync(any()))
+ .thenAnswer(
+ invocation -> {
+ DurableCallable callable = invocation.getArgument(0);
+ pending.set(worker.submit(callable::call));
+ assertThat(started.await(5, TimeUnit.SECONDS)).isTrue();
+ throw failure;
+ });
+ doAnswer(
+ invocation -> {
+ reportingStartedAt.set(Instant.now());
+ return null;
+ })
+ .when((ExecutionReporter) ctx)
+ .reportExecutionStartedAt(anyString(), anyString(), anyMap(), anyString());
+
+ try {
+ ToolCallAction.processToolRequest(
+ new ToolRequestEvent("test-model", List.of(toolCall("call-1"))), ctx);
+ assertReports(ctx, List.of("call-1"), List.of(), List.of("call-1"));
+ ArgumentCaptor finishedAt = ArgumentCaptor.forClass(String.class);
+ verify((ExecutionReporter) ctx)
+ .reportExecutionFailedAt(
+ anyString(),
+ anyString(),
+ anyMap(),
+ eq(failure),
+ anyString(),
+ finishedAt.capture());
+ assertThat(Instant.parse(finishedAt.getValue()))
+ .isBeforeOrEqualTo(reportingStartedAt.get());
+ assertBusinessResponsesFailed(ctx);
+ } finally {
+ release.countDown();
+ worker.shutdown();
+ assertThat(worker.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
+ }
+ assertThat(pending.get().get().isSuccess()).isTrue();
+ assertReports(ctx, List.of("call-1"), List.of(), List.of("call-1"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void parallelTimeoutTimestampPrecedesResponseProcessingAndReporting(
+ boolean completeDuringReporting) throws Exception {
+ TimeoutException failure = new TimeoutException("batch timed out");
+ CountDownLatch started = new CountDownLatch(2);
+ CountDownLatch release = new CountDownLatch(1);
+ ExecutorService workers = Executors.newFixedThreadPool(2);
+ List> pending = new ArrayList<>();
+ AtomicReference responseProcessingStartedAt = new AtomicReference<>();
+ ToolResponse firstResponse = mock(ToolResponse.class);
+ when(firstResponse.isSuccess())
+ .thenAnswer(
+ invocation -> {
+ responseProcessingStartedAt.compareAndSet(null, Instant.now());
+ return true;
+ });
+ Tool tool = mock(Tool.class);
+ when(tool.call(any()))
+ .thenAnswer(
+ invocation -> {
+ ToolParameters parameters = invocation.getArgument(0);
+ if ("call-1".equals(parameters.getParameter("query"))) {
+ return firstResponse;
+ }
+ started.countDown();
+ assertThat(release.await(5, TimeUnit.SECONDS)).isTrue();
+ return ToolResponse.success("late result");
+ });
+ RunnerContext ctx = parallelContext(tool);
+ doAnswer(
+ invocation -> {
+ List> callables =
+ invocation.getArgument(0);
+ ToolResponse response = callables.get(0).call();
+ pending.add(workers.submit(callables.get(1)::call));
+ pending.add(workers.submit(callables.get(2)::call));
+ assertThat(started.await(5, TimeUnit.SECONDS)).isTrue();
+ return List.of(
+ Outcome.success(response),
+ Outcome.failure(failure),
+ Outcome.failure(failure));
+ })
+ .when(ctx)
+ .durableExecuteAllAsync(any());
+ doAnswer(
+ invocation -> {
+ if (completeDuringReporting) {
+ release.countDown();
+ for (Future future : pending) {
+ future.get(5, TimeUnit.SECONDS);
+ }
+ }
+ return null;
+ })
+ .when((ExecutionReporter) ctx)
+ .reportExecutionStartedAt(anyString(), anyString(), anyMap(), anyString());
+
+ try {
+ ToolCallAction.processToolRequest(parallelRequest(), ctx);
+ ArgumentCaptor finishedAt = ArgumentCaptor.forClass(String.class);
+ verify((ExecutionReporter) ctx, times(2))
+ .reportExecutionFailedAt(
+ anyString(),
+ anyString(),
+ anyMap(),
+ eq(failure),
+ anyString(),
+ finishedAt.capture());
+ assertThat(finishedAt.getAllValues().get(0))
+ .isEqualTo(finishedAt.getAllValues().get(1));
+ assertThat(Instant.parse(finishedAt.getValue()))
+ .isBeforeOrEqualTo(responseProcessingStartedAt.get());
+ } finally {
+ release.countDown();
+ workers.shutdown();
+ assertThat(workers.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
+ }
+ assertReports(
+ ctx,
+ List.of("call-1", "call-2", "call-3"),
+ List.of("call-1"),
+ List.of("call-2", "call-3"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void parallelTimeoutOmitsStartsAfterResultObservation(boolean startsAfterObservation)
+ throws Exception {
+ TimeoutException failure = new TimeoutException("batch timed out");
+ Instant base = Instant.parse("2026-01-01T00:00:00Z");
+ Instant observedAt = base.plusSeconds(1);
+ Instant delayedStart = startsAfterObservation ? observedAt.plusSeconds(1) : observedAt;
+ AtomicReference now = new AtomicReference<>(base);
+ List> delayed = new ArrayList<>();
+ Tool tool = mock(Tool.class);
+ when(tool.call(any())).thenReturn(ToolResponse.success("ok"));
+ RunnerContext ctx = parallelContext(tool);
+ doAnswer(
+ invocation -> {
+ List> callables =
+ invocation.getArgument(0);
+ ToolResponse first = callables.get(0).call();
+ delayed.addAll(callables.subList(1, callables.size()));
+ now.set(observedAt);
+ return List.of(
+ Outcome.success(first),
+ Outcome.failure(failure),
+ Outcome.failure(failure));
+ })
+ .when(ctx)
+ .durableExecuteAllAsync(any());
+ doAnswer(
+ invocation -> {
+ Map metadata = invocation.getArgument(2);
+ if ("call-1"
+ .equals(metadata.get(ToolExecutionMetadataKeys.TOOL_CALL_ID))) {
+ // The delayed calls enter after the Action has observed the
+ // timeout.
+ now.set(delayedStart);
+ for (DurableCallable callable : delayed) {
+ callable.call();
+ }
+ }
+ return null;
+ })
+ .when((ExecutionReporter) ctx)
+ .reportExecutionStartedAt(anyString(), anyString(), anyMap(), anyString());
+
+ try (MockedStatic clock = mockStatic(Instant.class)) {
+ clock.when(Instant::now).thenAnswer(invocation -> now.get());
+ ToolCallAction.processToolRequest(parallelRequest(), ctx);
+ }
+
+ assertReports(
+ ctx,
+ startsAfterObservation ? List.of("call-1") : List.of("call-1", "call-2", "call-3"),
+ List.of("call-1"),
+ List.of("call-2", "call-3"));
+ verify(tool, times(3)).call(any());
+ ArgumentCaptor finishedAt = ArgumentCaptor.forClass(String.class);
+ verify((ExecutionReporter) ctx, times(2))
+ .reportExecutionFailedAt(
+ anyString(),
+ anyString(),
+ anyMap(),
+ eq(failure),
+ anyString(),
+ finishedAt.capture());
+ assertThat(finishedAt.getAllValues())
+ .containsExactly(observedAt.toString(), observedAt.toString());
+ ArgumentCaptor response = ArgumentCaptor.forClass(Event.class);
+ verify(ctx).sendEvent(response.capture());
+ assertThat(((ToolResponseEvent) response.getValue()).getSuccess())
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.of("call-1", true, "call-2", false, "call-3", false));
+ }
+
+ private static RunnerContext parallelContext(Tool tool) throws Exception {
+ RunnerContext ctx =
+ mock(RunnerContext.class, withSettings().extraInterfaces(ExecutionReporter.class));
+ when(ctx.getResource("search", ResourceType.TOOL)).thenReturn(tool);
+ when(ctx.getConfig()).thenReturn(toolCallConfig(true, 3));
+ when(ctx.durableExecuteAllAsync(any()))
+ .thenAnswer(
+ invocation -> {
+ List> callables =
+ invocation.getArgument(0);
+ List> outcomes = new ArrayList<>();
+ for (DurableCallable callable : callables) {
+ try {
+ outcomes.add(Outcome.success(callable.call()));
+ } catch (Exception e) {
+ outcomes.add(Outcome.failure(e));
+ }
+ }
+ return outcomes;
+ });
+ return ctx;
+ }
+
+ private static ToolRequestEvent parallelRequest() {
+ return new ToolRequestEvent(
+ "test-model", List.of(toolCall("call-1"), toolCall("call-2"), toolCall("call-3")));
+ }
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static void assertReports(
+ RunnerContext ctx, List started, List succeeded, List failed)
+ throws Exception {
+ ExecutionReporter reporter = (ExecutionReporter) ctx;
+ ArgumentCaptor