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

[joeyutong]

What changed

This PR implements the built-in operational metrics proposed in Discussion #901.

It builds on the execution lifecycle and trace context introduced by the merged #924. Metrics and Event Log recording consume the same in-process execution events independently; metrics are not derived by reading the Event Log.

Runtime lifecycle integration

• ActionExecutionOperator records input queue, input-run, Action task, and Action execution boundaries at the points where they occur.
• OperatorStateManager exposes pending Action state for restoring current-count gauges after task recovery.
• BuiltInMetrics is the central dispatcher. Action lifecycle events feed Action metrics, while LLM and Tool lifecycle events feed execution-entity metrics.

Metric implementations

• BuiltInInputRunMetrics records run outcomes, end-to-end, queue, and processing latency, pending input Events, and active input runs.
• BuiltInActionMetrics records scheduling and logical execution latency, pending Action tasks, and active Action executions.
• BuiltInExecutionMetrics pairs execution start and terminal events by execution id, then dispatches by entity type.
• LlmExecutionMetricRecorder records model-resource success, failure, and latency.
• ToolExecutionMetricRecorder records Tool metrics and projects explicit Skill and MCP Server metadata into independent scopes.
• Current-count gauges are rebuilt from Flink state after restore. Historical latency and outcome samples that began in a previous task attempt are intentionally not reconstructed.

Tool outcomes retain the existing language-specific contracts. Java maps an unsuccessful ToolResponse to failure. Python maps resource preparation and invocation exceptions to failure, while a normal arbitrary return remains successful because Python currently has no explicit error-result type. Strict alignment is tracked in #956 and is planned after the parallel Tool-call work in #926. This PR retains the Tool and MCP outcome counters and does not infer failure from arbitrary return payloads.

Java and Python ChatModel paths

• Each framework-observed ChatModel invocation emits its own LLM execution outcome and latency sample, including retry attempts.
• Retry count and wait time are recorded under the configured model_resource in both Java and Python, including final-failure and IGNORE paths.
• With Java model routing, retry metrics are attributed to the concrete candidate ChatModel resource used by each attempt.
• Existing request-scoped token metrics remain under the provider model scope.

Metric scope and documentation

• The Agent name is used as the Flink operator name, with the existing operator name retained as a fallback.
• The monitoring documentation lists the new scopes, metric types, lifecycle boundaries, retry semantics, recovery limitations, and current Tool outcome mapping.
• The configuration reference now describes retry metrics under the configured ChatModel resource rather than the underlying connection.

Validation

• mvn -T4 -B --no-transfer-progress spotless:check
• mvn -B --no-transfer-progress -pl plan,runtime -am -DskipITs -Dtest=ChatModelActionRetryTest,ChatModelActionRoutingTest,CompileUtilsTest,BuiltInActionMetricsTest,BuiltInExecutionMetricsTest,BuiltInInputRunMetricsTest,ActionExecutionOperatorTest -Dsurefire.failIfNoSpecifiedTests=false test
  • Plan: 27 tests passed
  • Runtime: 67 tests passed
• From python/: pytest -q flink_agents/plan/tests/actions/test_chat_model_action_retry.py flink_agents/api/tests/test_execution_reporter.py flink_agents/runtime/tests/test_flink_runner_context_trace.py
  • Python: 14 tests passed
• From python/: ruff check flink_agents/plan/actions/chat_model_action.py flink_agents/plan/tests/actions/test_chat_model_action_retry.py

Related work

• Agent Trace implementation: #924
• Agent Trace proposal: Discussion #900
• Operational Metrics proposal: Discussion #901
• Parallel Tool-call execution: #925, #926
• Java/Python Tool outcome alignment: #956

[wenjin272]

Thanks for putting together this comprehensive metrics implementation and the accompanying tests. I left three inline comments for issues that should be addressed.

[wenjin272]

toolName comes directly from the LLM-generated tool call, and ToolCallAction reports lifecycle events even when resource lookup fails. Therefore, every hallucinated or invalid tool name creates a new tool=<name> metric group here. Since these groups are not removed, arbitrary model output can cause unbounded metric cardinality and eventually exhaust the task or metrics backend.

Please only create per-tool groups for configured tool names and aggregate unknown names into a fixed bucket such as unknown.

[joeyutong]

Fixed. Per-Tool metrics now consult the runtime resource registry; unregistered names are aggregated under tool=unknown, while Agent Trace keeps the requested name. Added a regression test covering multiple invalid names.

[wenjin272]

After recovery, restoreActionTask() increments activeActionExecutions for a restored task whose execution had already started. If durable state shows that the action completed before the failure, the operator emits execution_reused instead of execution_finished or execution_failed.

This branch ignores that event, so the restored execution remains in activeExecutions and numOfActiveActionExecutions never returns to zero. Please handle execution_reused as a terminal event and add a restore-to-reused regression test. No latency sample needs to be recorded when the restored start timestamp is unavailable.

[joeyutong]

Fixed. execution_reused now closes a restored active Action execution without recording latency when the original start timestamp is unavailable. Added a restore-to-reused regression test.

[wenjin272]

markInputEventReceived() has already stored timing for this event, but resolveContextKey() can throw, for example while converting a Python key. Because this call runs before markInputRunStarted() and outside both failure handlers, such an input is never counted as failed and its entry remains in receivedInputNanos until restart.

Please catch failures during context-key and trace-context creation and call markInputEventFailed(inputEvent), or include this setup in a broader failure boundary around processInputEvent().

[joeyutong]

Fixed. Context-key and trace-context setup failures now close the received Input Event via markInputEventFailed before an Input Run is started. Added an operator regression test for key-conversion failure.

[weiqingy]

Thanks for taking this on. A few questions inline.

[weiqingy]

skillName here comes straight from the model's tool arguments (LoadSkillTool.java:72-74, skill_tools.py:93-95). Nothing checks it against the registry.

The unknown bucketing at :62-68 works, but it cannot reach this line. load_skill is itself a registered ResourceType.TOOL (AgentPlan.java:614-620), so the tool key resolves fine and the raw string flows into the skill scope just below. Each new value allocates a sub-group, a counter and a DescriptiveStatisticsHistogram(100) (FlinkAgentsMetricGroupImpl.java:105-110) for the life of the TaskManager, plus a Prometheus label. One made-up skill name per run is enough to grow it without bound. monitoring.md:83 says cardinality is bounded, which is true for tool.

Both runtimes already have the registry to hand (LoadSkillTool.java:102, skill_tools.py:121), and the Trace record keeps the requested name either way. Would validating inside getToolExecutionMetadata be the cleaner spot, or would you rather pass a second predicate next to isRegisteredTool?

[joeyutong]

Fixed. Java and Python now record whether the requested Skill is registered. Registered names keep their own scope; all other names are aggregated under skill=unknown, while Agent Trace retains the requested name. Regression coverage was added.

[weiqingy]

This counter goes up whatever the outcome, and it can never see a failed one. LoadSkillTool returns ToolResponse.success(...) on all seven paths and never calls ToolResponse.error. Two of those paths are errors: skill not found at :105, resource not found at :143. Python does the same at skill_tools.py:116, :125, :151.

So load_skill(name="does-not-exist") raises numOfToolCallsSucceeded{tool=load_skill}, raises numOfSkillLoads{skill=does-not-exist}, and records a latency sample. No failure shows up anywhere.

Returning an error would not cost the model the "Available skills: ..." hint, since the failure branch already puts response.getError() into the TOOL message (ChatModelAction.java:661-664). This also looks separate from #956, which is about Python having no error-result type. Here Java has ToolResponse.error and uses it nowhere. Should the not-found paths return it, with a matching change on the Python side?

[joeyutong]

Kept the existing Tool outcome contract in this PR. numOfSkillLoads is now documented as counting terminal load_skill calls regardless of outcome, including the current not-found normal-return behavior. Explicit failure-result alignment is tracked in #956.

[weiqingy]

Makes sense, and #956 is the right home for the error-result change.

[weiqingy]

This teaches the Action scope about replay. The LLM and Tool scopes have no equivalent.

durableExecuteCompletionOnly returns a cached result at RunnerContextImpl.java:552 without calling the callable, but the reports around it fire either way. ChatModelInvoker.java:157 starts and :174-175 succeeds around durableExecute at :162-163, and ToolCallAction.java:112-113 / :198-199 has the same shape. So a replay looks like a start and a finish microseconds apart: numOfLlmCallsSucceeded goes up and llmCallLatencyMs takes a ~0 ms sample for a call that never ran, which is not what monitoring.md:81 promises. The isCompleted() shortcut at ActionExecutionOperator.java:468 only catches the case where the whole Action is complete, not the recovery where just the inner call is cached.

This only reaches jobs that turn on an action state store (ACTION_STATE_STORE_BACKEND defaults to null, AgentConfigOptions.java:65-66). One knock-on if you go this way: BuiltInExecutionMetrics returns on any non-terminal type at :76-78, so a reused event would need to drain activeExecutionStartNanos there too. Is emitting executionReused() on the cache hit, handled the way you handle it here, the direction you would take?

[joeyutong]

Agreed. This is the existing Agent Trace durable-replay limitation: child cache reuse is not exposed, so these metrics currently inherit the fresh-success and near-zero-latency behavior during fine-grained recovery. I documented the metric impact explicitly; distinguishing reused child executions remains follow-up work.

[weiqingy]

Two things are riding on these rows.

The scope key and the value both moved. It was getSubGroup("model", chatModel.getConnectionName()) before this PR and is getSubGroup("model_resource", modelResource) now (ChatModelAction.java:205-217, and _record_retry_metrics matches). The resource name is the better identity. But every existing retryCount and retryWaitSec series breaks, and neither doc mentions it.

The ErrorHandlingStrategy.RETRY qualifier the old rows carried is also gone (grep -c ErrorHandlingStrategy monitoring.md returns 0), while configuration.md:133, edited in this same PR, still has it. The gate has not changed and is off by default (AgentExecutionOptions.java:24-28, ChatModelAction.java:383-394), so on a default config both metrics stay at zero and only one of the two docs explains why.

Row 70's "Only recorded when at least one retry occurs" half covers it, and retryWaitSec has nothing. Worth putting the qualifier back on both rows and adding a line about the scope move?

[joeyutong]

Updated the docs. Both retry metrics now state the ErrorHandlingStrategy.RETRY gate, and the migration from model.<connection_name> to model_resource.<resource_name> is called out explicitly.

[weiqingy]

monitoring.md:164 documents the prefix format, and this PR adds that <operator_name> is the agent name, so the end state is written down. What no doc says is that existing series move. Everything this operator publishes used to sit under action-execute-operator and now sits under the agent name, so custom agent_metric_group and action_metric_group metrics break, along with any dashboard keyed on the old prefix. Your own TokenMetricsE2ETest edit shows how far it reaches: a token metric's expected prefix had to change.

The PR body calls this "the existing operator name retained as a fallback", which reads gentler than what it means for anyone already scraping these. How would you want to flag the rename, a line in monitoring.md or a release note?

[joeyutong]

Updated monitoring.md with the migration behavior: operator_name now uses the Agent name, falls back to action-execute-operator only when unavailable, and dashboards filtering the old value must be updated. The metric hierarchy is unchanged.

[weiqingy]

Neither Math.max is exercised by the new tests. BuiltInActionMetricsTest never dequeues while the pending count is already 0. duplicateTerminalNotificationDoesNotUnderflowActiveGauge looks like the coverage, but the idempotency it proves sits a level up: activeInputRunIds.remove(inputRunId) short-circuits on the second call (BuiltInInputRunMetrics.java:145), so decrement() at :151 never runs. restoreActiveInputRuns also pre-clamps at :132 before it calls set.

I dropped both Math.max calls and the runtime suite still passed 688/688. So nothing in the ~880 new test lines tells a clamped gauge apart from an unclamped one.

Two assertions would close it: decrement() at 0 stays 0, and set(-1) gives 0. Where would those sit best, a small CurrentCountGaugeTest, or an unbalanced sequence routed through the higher-level classes?

[joeyutong]

Added a focused CurrentCountGaugeTest covering decrement() at zero and set(-1), so both clamping branches are exercised directly.

[weiqingy]

actionMetricGroups is built once from the current plan at :83-89. But restoreActionTask reaches this lookup at :151-155 with a name read from a deserialized ActionTask (ActionExecutionOperator.java:845-853). The live callers all take their names from the current plan, so the strict throw fits them. The restore path is the odd one out.

I could not build a restore that actually hits this, so it is a question rather than a claim. The case I have in mind is a savepoint holding in-flight action tasks, restored after an action was renamed or removed. tryResumeProcessActionTasks runs inside open() (:239), so the throw would fail operator startup rather than just lose a metric, and skipping an unknown action or logging once looks like about three lines. Does the state format tolerate a plan change across a restore today?

[joeyutong]

Updated the metric restore path so it no longer throws solely because a restored Action is absent from the current Plan; its metric group is created lazily. Regular live Action lookups remain strict, and a restored-action regression test was added.

[weiqingy]

Thanks for working through these. All seven from the last round look addressed to me. Three things I held back last time are inline.

[weiqingy]

activeExecutionStartNanos is filled at :67 and drained only by a matching EXECUTION_FINISHED or EXECUTION_FAILED at :81. No cap and no sweep, and the object is built once per operator (BuiltInMetrics.java:83-84), so a start that never gets its terminal stays for the operator's lifetime.

The map one layer up is swept. RunnerContextImpl.java:417-430 pairs the same starts and terminals in activeReportedExecutions, and ActionTaskContextManager.java:306-308 drops that map per action execution. The same orphan is cleaned up there and kept here.

Framework code still looks balanced on every path, so this needs a user action to trigger. ExecutionReporters is public and dispatches on ctx instanceof ExecutionReporter (ExecutionReporters.java:99-101), so an action that reports an llm or tool start and no terminal leaks one entry per call.

Is the operator-lifetime map deliberate, or should it be swept when the action execution completes?

[joeyutong]

Good catch. Child execution start timestamps are now grouped by their parent Action execution and the group is removed when that Action reaches finished, failed, or reused. The new test verifies that cleanup is scoped to the terminated Action and preserves latency state for another active Action.

[weiqingy]

These four names land in the same group users get for their own metrics. RunnerContextImpl.java:248-250 returns agentMetricGroup.getSubGroup("action", actionName), the same instance BuiltInMetrics.java:190 hands to BuiltInActionMetrics, and FlinkAgentsMetricGroupImpl.java:77-82 returns the cached metric when the name already exists. So ctx.getActionMetricGroup().getGauge("numOfPendingActionTasks") returns the framework gauge, and whichever writer runs last wins.

monitoring.md:106 points users at this group for custom metrics, and this PR adds four reserved names to it. Is one sentence reserving the built-in names enough, or would you rather the group reject a collision outright?

[joeyutong]

Agreed. I documented built-in metric names as reserved in their corresponding scopes, both in the RunnerContext Javadocs and the monitoring guide. I kept runtime rejection out of this PR because framework and user metrics currently share the same get-or-create API; enforcing ownership needs a separate registration boundary.

[weiqingy]

nit: this defines a registered tool as a key in agentPlan.getResourceProviders().get(ResourceType.TOOL). Production passes something wider: toolName -> resourceCache.hasResource(toolName, ResourceType.TOOL) (ActionExecutionOperator.java:193-197), and ResourceCache.java:101-108 checks the cache first, so it also matches resources put in with no provider.

Both callers of this constructor are tests (EventRouterTest.java:291, BuiltInMetricsTest.java:39), so the definition that is easiest to read is the one production never runs. BuiltInMetrics has no access to the cache, so is dropping the convenience constructor and letting the two tests pass their own predicate the simpler end state?

[joeyutong]

Agreed. I removed the two-argument constructor and updated both tests to pass an explicit predicate. Production now remains the only source of Tool registration semantics through ResourceCache.hasResource.

[weiqingy]

Thanks for working through these. All three are closed on my side. One small thing about the new test inline.

[weiqingy]

nit: both executions land on the same group path, restored_action -> primary_model (:76-82), so :98-104 counts one shared histogram. It catches no sweep (2) and over-sweep (0), but not which group got swept. Dropping activeAction's instead of completedAction's also gives 1.

Would renaming the second child to secondary_model let you assert 0 on the completed one and 1 on the active one?

[joeyutong]

Good catch. I split the child executions into primary_model and secondary_model, then assert 0 for the completed Action and 1 for the still-active Action. This now pins which Action-scoped state was removed.

[weiqingy]

Thanks for addressing the comments. The rebase pulled in parallel tool calls, which leaves one remaining issue. Inline below.

[weiqingy]

llmCallLatencyMs at :69 says exactly what its window covers. This row says only "Tool call latency", and the window is wider than one tool: ToolCallAction.java:136 reports started for every tool before any runs, and the terminals all fire after the batch returns (:205-208). So a fast tool in a batch reports the batch's time, and below JDK 21, where the batch runs serially (ContinuationActionExecutor.java:76-95), it also carries the tools ahead of it. skillLoadLatencyMs and mcpToolCallLatencyMs share the value, and Python matches (tool_call_action.py:169, :219-220).

The cause sits upstream in ToolCallAction, so the doc may be the only lever in this PR. The tool.<tool_name> scope reads like a promise that you can spot the slow tool in a batch. Is that what you want it to carry, or is the batch number the one worth publishing?