This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 47daff57b [docs] Improve consistency and configuration guidance (#1284)
47daff57b is described below
commit 47daff57b5c4189c80dd7037660f2a7832270879
Author: Yu Tang <[email protected]>
AuthorDate: Tue Oct 20 22:30:35 2026 +0800
[docs] Improve consistency and configuration guidance (#1284)
Generated-by: Codex CLI 0.163.0-alpha.10.2 (GPT-6)
Co-Authored-By: Codex <[email protected]>
---
docs/content/_index.md | 2 +-
docs/content/docs/development/calling-llms/chat.md | 5 +-
.../docs/development/calling-llms/model_routing.md | 4 +-
.../docs/development/calling-llms/tool_use.md | 4 +-
.../docs/development/create-agent/actions.md | 30 +-
.../docs/development/memory/memory_events.md | 2 +-
.../docs/development/multi-agent/subagents.md | 8 +-
docs/content/docs/get-started/overview.md | 2 +-
docs/content/docs/get-started/setup.md | 5 +-
docs/content/docs/operations/configuration.md | 404 ++++++++++-----------
docs/content/docs/operations/consistency.md | 356 ++++++++++++++++++
docs/content/docs/operations/deployment.md | 61 +---
docs/content/docs/operations/monitoring.md | 123 ++++++-
13 files changed, 692 insertions(+), 314 deletions(-)
diff --git a/docs/content/_index.md b/docs/content/_index.md
index 06d239970..5e8d316a1 100644
--- a/docs/content/_index.md
+++ b/docs/content/_index.md
@@ -45,7 +45,7 @@ It provides a step by step introduction on how to install
Flink Agents and build
### How to Operate an Agent
-* [Deployment]({{< ref "docs/operations/deployment" >}}), [Configuration]({{<
ref "docs/operations/configuration" >}}), and [Monitoring]({{< ref
"docs/operations/monitoring" >}}) - Deploy, configure, and monitor agents
+* [Deployment]({{< ref "docs/operations/deployment" >}}), [Consistency]({{<
ref "docs/operations/consistency" >}}), [Configuration]({{< ref
"docs/operations/configuration" >}}), and [Monitoring]({{< ref
"docs/operations/monitoring" >}}) - Deploy, configure, recover, and monitor
agents
<--->
diff --git a/docs/content/docs/development/calling-llms/chat.md
b/docs/content/docs/development/calling-llms/chat.md
index e86388fff..102278405 100644
--- a/docs/content/docs/development/calling-llms/chat.md
+++ b/docs/content/docs/development/calling-llms/chat.md
@@ -259,13 +259,12 @@ the failure.
<span id="failures-and-recovery"></span>
Model-call failures are reported after configured retries and routing fallbacks
-are exhausted. See [Core Options]({{< ref
"docs/operations/configuration#core-options" >}})
+are exhausted. See [Model calls, tools, and retries]({{< ref
"docs/operations/configuration#model-calls-tools-and-retries" >}})
for retry settings and [Model Routing]({{< ref
"docs/development/calling-llms/model_routing" >}})
for fallbacks.
For how calls behave after a failure and recovery, see
-[Recovery Behavior]({{< ref
"docs/development/create-agent/actions#recovery-behavior" >}})
-in Actions.
+[Recovery flow]({{< ref "docs/operations/consistency#recovery-flow" >}}).
## Multimodal Messages {#message-content-blocks}
diff --git a/docs/content/docs/development/calling-llms/model_routing.md
b/docs/content/docs/development/calling-llms/model_routing.md
index 58db5d44b..c59efbc4f 100644
--- a/docs/content/docs/development/calling-llms/model_routing.md
+++ b/docs/content/docs/development/calling-llms/model_routing.md
@@ -281,7 +281,7 @@ behaviors:
For candidates `[small, medium, big]`, selecting `big` with fallback enabled
tries
`big`, then `small`, then `medium`. Model-call retries run before fallback; see
-[Core Options]({{< ref "docs/operations/configuration#core-options" >}}).
+[Model calls, tools, and retries]({{< ref
"docs/operations/configuration#model-calls-tools-and-retries" >}}).
Failures before the call, such as an unresolved model resource, can proceed
directly to the next candidate without retries.
@@ -354,7 +354,7 @@ metadata when distinguishing their usage.
With durable execution enabled, completed routing decisions and model calls can
be reused after recovery. Without it, a request may be routed again and select
a
different model. See
-[Recovery Behavior]({{< ref
"docs/development/create-agent/actions#recovery-behavior" >}})
+[Consistency]({{< ref
"docs/operations/consistency#recovery-guarantees-and-user-responsibilities" >}})
for the persistence and replay guarantees.
## Examples
diff --git a/docs/content/docs/development/calling-llms/tool_use.md
b/docs/content/docs/development/calling-llms/tool_use.md
index ce33ef845..dde007f38 100644
--- a/docs/content/docs/development/calling-llms/tool_use.md
+++ b/docs/content/docs/development/calling-llms/tool_use.md
@@ -437,9 +437,9 @@ inside each tool, such as an HTTP client read timeout.
A batch timeout does not undo external effects or guarantee that an already
running tool stops. Tools that change external systems should be idempotent,
because a call may run again after recovery. See
-[Configuration]({{< ref "docs/operations/configuration" >}}) for the full
async batch
+[Configuration]({{< ref
"docs/operations/configuration#runtime-and-async-execution" >}}) for the full
async batch
and execution-capacity options, and
-[Durable Execution]({{< ref
"docs/development/create-agent/actions#durable-execution" >}})
+[Consistency]({{< ref
"docs/operations/consistency#recovery-guarantees-and-user-responsibilities" >}})
for recovery guarantees.
## Tool response content and metadata
diff --git a/docs/content/docs/development/create-agent/actions.md
b/docs/content/docs/development/create-agent/actions.md
index 097ae9e92..adcc38e33 100644
--- a/docs/content/docs/development/create-agent/actions.md
+++ b/docs/content/docs/development/create-agent/actions.md
@@ -562,8 +562,9 @@ For example, if an Action fails after a profile lookup has
completed and its res
been persisted, recovery can reuse that profile instead of querying the
service again.
Code outside the durable call may still run again when the Action is
re-executed.
-Durable execution requires an external action state store. Configure it as
described in
-[Exactly-Once Action Consistency]({{< ref
"docs/operations/deployment#exactly-once-action-consistency" >}}).
+To save and reuse call outcomes during recovery, configure checkpointing and
an external
+Action State Store as described in
+[Consistency]({{< ref
"docs/operations/consistency#configure-an-action-state-store" >}}).
#### A Durable Call
@@ -628,18 +629,10 @@ Java's `getResultClass()` provides the type needed to
deserialize a saved result
#### Recovery Behavior
-When recovery re-executes an unfinished Action and reaches a durable call:
-
-| State of the matching call | Recovery behavior |
-| --- | --- |
-| A successful result has been persisted | Return the saved result without
calling the function again. |
-| A failure has been persisted | Replay the failure without calling the
function again. |
-| No completed outcome has been persisted | Execute the function again, or use
the reconciler for a previously recorded in-flight call. |
-
-An external operation may finish before its result is persisted. If a failure
occurs in
-that interval, calling the operation again can repeat its side effects. For
such
-operations, use the external service's idempotency support or a reconciler
that can
-recover the existing outcome.
+See [Recovery flow]({{< ref "docs/operations/consistency#recovery-flow" >}})
for how recovery
+handles saved outcomes and in-flight calls, and
+[Limits on external side effects]({{< ref
"docs/operations/consistency#limits-on-external-side-effects" >}})
+for operations that may repeat.
#### Reconciling an In-flight Call
@@ -799,9 +792,10 @@ worker and batch limits. Concurrency between Actions has
additional requirements
the cross-language Flink versions below.
- **Python Actions:** use `async def` and the framework's futures and `gather`
APIs.
-See [Core Options]({{< ref "docs/operations/configuration#core-options" >}})
for worker-pool
-and concurrency limits. These scheduling requirements apply to both ordinary
and durable
-async calls. Durable calls additionally require the configured action state
store.
+See [Runtime and async execution]({{< ref
"docs/operations/configuration#runtime-and-async-execution" >}})
+for worker-pool and concurrency limits. These scheduling requirements apply to
both ordinary
+and durable async calls. Saving durable call outcomes for recovery
additionally requires
+an Action State Store.
#### Cross-language Async Requirements
@@ -886,7 +880,7 @@ missing value.
Reading a missing attribute without a guard or applying an operation to an
incompatible
value can fail while evaluating a condition for an Event. For failure handling
options, see
-[`action.trigger-condition.evaluate-failure-strategy`]({{< ref
"docs/operations/configuration#core-options" >}})
+[`action.trigger-condition.evaluate-failure-strategy`]({{< ref
"docs/operations/configuration#condition-handling" >}})
in Configuration.
#### Limitations
diff --git a/docs/content/docs/development/memory/memory_events.md
b/docs/content/docs/development/memory/memory_events.md
index fd5bc9c14..49b08493e 100644
--- a/docs/content/docs/development/memory/memory_events.md
+++ b/docs/content/docs/development/memory/memory_events.md
@@ -238,7 +238,7 @@ Because `agent-run.begin-event` is disabled by default, the
Trace Log has no bui
## Semantics & Caveats
-- **Trace Log level.** Configure [recording targets]({{< ref
"docs/operations/configuration#trace-log-targets" >}}) with `scope: EVENT_ONLY`
and `level: VERBOSE` when full memory-event attributes are required, or select
individual Event names with an entity scope. Memory-event generation and log
selection are independent: the emission switches determine whether Events are
produced, while targets determine whether they are written to Trace Log.
+- **Trace Log level.** Configure [recording targets]({{< ref
"docs/operations/monitoring#recording-targets" >}}) with `scope: EVENT_ONLY`
and `level: VERBOSE` when full memory-event attributes are required, or select
individual Event names with an entity scope. Memory-event generation and log
selection are independent: the emission switches determine whether Events are
produced, while targets determine whether they are written to Trace Log.
- **Durable execution.** A long-term memory operation wrapped in
`durable_execute` is not recorded again when the durable result is replayed
from state. The generated events describe operations that actually executed in
the current attempt.
- **Observation failures.** Unsupported short-term or sensory values are
omitted. Invalid long-term records are skipped individually after a valid batch
is parsed, but a bridge, batch-serialization, or malformed-payload failure
drops the current partition key's LTM observation batch. Warnings do not
include observed values. These failures do not fail the action or roll back a
completed long-term backend operation.
- **Failed actions.** A failed action does not reach the memory-event emission
boundary. The operator task fails and recovery rebuilds the interpreter, so
queued actions do not continue in the same interpreter instance after the
failure.
diff --git a/docs/content/docs/development/multi-agent/subagents.md
b/docs/content/docs/development/multi-agent/subagents.md
index 0472561db..b207ba9fa 100644
--- a/docs/content/docs/development/multi-agent/subagents.md
+++ b/docs/content/docs/development/multi-agent/subagents.md
@@ -1253,8 +1253,12 @@ triggers recovery.
Sub-agent calls use
[durable execution]({{< ref
"docs/development/create-agent/actions#durable-execution" >}}).
-Recovery reuses persisted results. Work whose result has not been persisted may
-run again, so external services need to handle retried requests.
+With checkpointing and an
+[Action State Store]({{< ref
"docs/operations/consistency#configure-an-action-state-store" >}}),
+recovery reuses persisted results. Work whose result has not been persisted may
+run again, so external services need to handle retried requests. See
+[Consistency]({{< ref
"docs/operations/consistency#recovery-guarantees-and-user-responsibilities" >}})
+for replay conditions and limits on repeated effects.
The framework generates stable `session_id` and `call_id` values for recovery.
If you supply your own IDs, keep them unchanged when the same call is retried.
diff --git a/docs/content/docs/get-started/overview.md
b/docs/content/docs/get-started/overview.md
index 9d1934640..24cb56785 100644
--- a/docs/content/docs/get-started/overview.md
+++ b/docs/content/docs/get-started/overview.md
@@ -33,7 +33,7 @@ Like any Agent OS, it manages the core building blocks of an
agent - orchestrati
The key features of Apache Flink Agents include:
- **Massive Scale and Millisecond Latency**: Processes massive-scale event
streams in real time, leveraging Flink's distributed processing engine.
- **Seamless Data and AI Integration**: Agents interact directly with Flink's
DataStream and Table APIs for input and output, enabling a smooth integration
of structured data processing and semantic AI capabilities within Flink.
-- **Reliable and Exactly-Once Execution**: Ensures exactly-once consistency
for agent actions and their side effects by integrating Flink's checkpointing
with an external write-ahead log, and supports durable execution - optionally
with reconcilers to reconcile in-flight side effects upon failure recovery - so
agents run reliably without human supervision.
+- **Reliable Execution**: Builds on Flink checkpointing and supports an
external Action State Store to reuse saved Action completions and durable call
outcomes during recovery. See [Consistency]({{< ref
"docs/operations/consistency" >}}) for setup, guarantees, and limits on
external side effects.
- **Familiar Agent Abstractions**: Leverages well-known AI agent concepts -
skills, short/long-term memory, prompts, tools, and dynamic orchestration -
making it easy for developers experienced with agent-based systems to quickly
adopt and build on Apache Flink Agents without a steep learning curve.
- **Multi-Language Support**: Provides Java, Python, and YAML APIs for
defining Agents. Actions and supported Resources can be implemented in Java or
Python and used across languages. See [Choose APIs and Languages]({{< ref
"docs/development/create-agent/choose-apis-and-languages" >}}).
- **Rich Ecosystem**: Natively integrates mainstream LLMs, vector stores from
diverse providers, and tools or prompts hosted on MCP servers into your agents,
while enabling customizable extensions.
diff --git a/docs/content/docs/get-started/setup.md
b/docs/content/docs/get-started/setup.md
index 80a61fb9a..f9eb97b7e 100644
--- a/docs/content/docs/get-started/setup.md
+++ b/docs/content/docs/get-started/setup.md
@@ -434,5 +434,6 @@ submits a complete example and shows its output:
- [Use the Built-in ReAct Agent]({{< ref
"docs/get-started/quickstart/react_agent" >}})
- [Define an Agent with YAML]({{< ref "docs/get-started/quickstart/yaml_agent"
>}})
-For a remote cluster, and for the consistency guarantees that apply once the
job runs
-there, see [Deployment]({{< ref "docs/operations/deployment" >}}).
+For submission to a remote cluster, see
+[Deployment]({{< ref "docs/operations/deployment#submit-to-flink-cluster" >}}).
+For consistency guarantees, see [Consistency]({{< ref
"docs/operations/consistency" >}}).
diff --git a/docs/content/docs/operations/configuration.md
b/docs/content/docs/operations/configuration.md
index d30592db4..1fd208f5f 100644
--- a/docs/content/docs/operations/configuration.md
+++ b/docs/content/docs/operations/configuration.md
@@ -1,6 +1,6 @@
---
title: Configuration
-weight: 2
+weight: 3
type: docs
---
<!--
@@ -24,16 +24,12 @@ under the License.
## How to configure Flink Agents
-There are two ways to configure Flink Agents, listed in order of **priority
from high to low**:
+Set options through `AgentsExecutionEnvironment` or the `agent` section of a
+Flink YAML configuration file. Values set on the environment override values
loaded
+from YAML.
-1. **Setting via the AgentsExecutionEnvironment**
-2. **Setting via a Flink YAML configuration file**
-
-The AgentsExecutionEnvironment applies to Agents from the
AgentsExecutionEnvironment, and the Flink YAML configuration file applies to
all Flink Agents Jobs using the same configuration file.
-
-{{< hint info >}}
-In case of duplicate keys, the value from the highest priority will override
those from lower priorities.
-{{< /hint >}}
+Environment settings apply to Agents added to that environment. YAML settings
apply
+to jobs that load the same file.
### Setting via the AgentsExecutionEnvironment
@@ -50,8 +46,7 @@ agents_env =
AgentsExecutionEnvironment.get_execution_environment(env)
# Get configuration object from the environment
config = agents_env.get_config()
-# Set custom configuration using a direct key (string-based key)
-# This is suitable for user-defined or non-standardized settings.
+# Set an option using its string key
config.set_int("kafkaActionStateTopicNumPartitions", 128)
# Set framework-level configuration using a predefined ConfigOption class
@@ -70,7 +65,7 @@ AgentsExecutionEnvironment agentsEnv =
AgentsExecutionEnvironment.getExecutionEn
// Get configuration object
Configuration config = agentsEnv.getConfig();
-// Set custom configuration using key (direct string key)
+// Set an option using its string key
config.setInt("kafkaActionStateTopicNumPartitions", 128); // Kafka topic
partitions count
// Set the list of event listeners
@@ -85,13 +80,11 @@ config.set(AgentExecutionOptions.MAX_RETRIES, 3);
### Setting via the Flink YAML configuration file
-The former `ErrorHandlingStrategy` API and `error-handling-strategy` option
have been removed. To enable retries, set `max-retries` to a positive number
(for example, 3 to retain the former `RETRY` mode's default budget). Its
default is now 0, preserving the previous default behavior of making no
retries. Terminal Chat failures are reported through failed `ChatResponseEvent`
events instead of a `FAIL` or `IGNORE` policy.
-
Flink Agents allows reading configurations from the Flink YAML configuration
file.
#### Format
-As part of the Flink configuration file, the flink agents configuration must
follow this format, with all agent-specific settings nested under the `agent`
key:
+Nest Flink Agents settings under the `agent` key:
```yaml
agent:
@@ -103,261 +96,234 @@ agent:
#### Loading Behavior
-By default, the configuration is automatically loaded from
`$FLINK_HOME/conf/config.yaml`.
-
-**Special Condition**
+When creating an `AgentsExecutionEnvironment` for a Flink job, Flink Agents
loads
+the `agent` section from the configuration directory specified by
`FLINK_CONF_DIR`.
+Set this environment variable before creating the environment:
-In the following case, Flink Agents may not locate the corresponding
configuration file, necessitating manual configuration. If the file is not set,
no configuration file will be loaded, potentially resulting in unexpected
behavior or failures.
-
-- **For MiniCluster**:
- Manual setup is **required** — always export the environment variable before
running the job:
+```bash
+export FLINK_CONF_DIR="/path/to/flink/conf"
+```
- ```bash
- export FLINK_CONF_DIR="path/to/your/config.yaml"
- ```
+`FLINK_CONF_DIR` points to a directory. Java reads `config.yaml` from that
directory.
+Python reads `flink-conf.yaml` if it exists, otherwise `config.yaml`.
- This ensures that Flink can locate and load the configuration file correctly.
+If `FLINK_CONF_DIR` is unset, no YAML configuration file is loaded.
Configuration
+through `AgentsExecutionEnvironment` still works and overrides values loaded
from
+the file.
## Built-in configuration options
-### Core Options
-Here is the list of all built-in core configuration options.
-
-| Key | Default | Type
| Description
|
-|---------------------------|----------------------------|-----------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `event-listeners` | none | `List<String>`
| The list of event listener class names. Each class must implement the
EventListener interface and provide a public no-argument constructor. {{< hint
warning >}} Note: Currently, custom event listeners are only supported in Java.
{{< /hint >}} |
-| `action.trigger-condition.evaluate-failure-strategy` | `WARN_AND_SKIP` |
ConditionEvaluationFailureStrategy | Handles failures while evaluating a
trigger condition for an individual Event, including missing attributes,
incompatible values, or a non-Boolean result. <br/><ul><li>`WARN_AND_SKIP`
(default): log a warning, treat that condition as false, and continue with
later OR conditions.</li><li>`FAIL`: throw `IllegalStateException` and fail the
Flink task; recovery follows the job's re [...]
-| `max-retries` | 0 | int
| Number of additional attempts per model call, including routing judge
calls. Defaults to 0 (no retries).
|
-| `retry-wait-interval` | 1 | int
| Base wait interval in seconds between retries. 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. The built-in
chat and decision actions report retry count and total wait time in
`ChatResponseEvent` and `DecisionResponseEvent`, respectively, and record
metrics (`retryCount`, `retryWaitSec`) un [...]
-| `chat.async` | true | boolean
| Whether chat asynchronously for built-in chat action.
|
-| `decision.async` | true | boolean
| Whether the built-in decision action invokes the model through durable
async execution. Java calls through Python fall back to synchronous execution
when the Flink runtime does not support asynchronous Python calls. |
-| `tool-call.async` | true | boolean
| Whether the built-in tool-call action runs each tool via durable async
execution. See [Parallel tool-call batches]({{< ref
"docs/development/calling-llms/tool_use#parallel-tool-call-batches" >}}) for
how this switch works with the batch settings below. |
-| `async.batch.parallelism` | os cpu count | int | Maximum concurrent calls in
one async batch, including batches created with `ctx.gather(...)`. `1` runs
calls sequentially. Actual concurrency is also limited by `num-async-threads`.
See [Parallel tool-call batches]({{< ref
"docs/development/calling-llms/tool_use#parallel-tool-call-batches" >}}). |
-| `async.batch.timeout.ms` | -1 (disabled) | long (milliseconds) | Overall
timeout for one async batch. Non-positive values disable it. A timeout does not
guarantee that running calls stop or undo their external effects. See [Timeouts
and external effects]({{< ref
"docs/development/calling-llms/tool_use#timeouts-and-external-effects" >}}) for
tool-call behavior and constraints. |
-| `rag.async` | true | boolean
| Whether retrieve context asynchronously for built-in context retrieval
action.
|
-| `num-async-threads` | os cpu count * 2 | int | Controls execution capacity
for asynchronous calls. `async.batch.parallelism` separately limits concurrency
within each batch. See [Parallel tool-call batches]({{< ref
"docs/development/calling-llms/tool_use#parallel-tool-call-batches" >}}) for
tool-call concurrency guidance. |
-| `parallel-execution.enabled` | true | boolean
| Whether plans with only **Java Actions** run Actions concurrently on the
parallel execution engine when the continuation-based engine is unavailable
(**JDK < 21**). Plans declaring Python Resources additionally require a Flink
version supporting [cross-language async calls]({{< ref
"docs/development/create-agent/actions#cross-language-async-requirements" >}}).
Same-key input records commit in input order, [...]
-| `max-in-flight-input-records` | 100 | int
| Maximum number of input records that may be in flight concurrently. Only
enforced by the **JDK < 21** parallel execution engine for plans with only
**Java Actions**; the JDK 21 coroutine engine and plans containing **Python**
actions ignore it. Every admitted record consumes one unit of budget; at the
cap, admission of further records blocks until an in-flight record retires. |
-| `job-identifier` | none | String
| The unique identifier of job, remaining consistent after restoring from a
savepoint. If not set, uses flink job id.
|
-| `short-term-memory.state-ttl.ms` | 0 | long
| Time-to-live for short-term memory state in milliseconds. Set to a value
greater than 0 to enable TTL; 0 disables it.
|
-| `short-term-memory.state-ttl.update-type` | `ON_READ_AND_WRITE` |
ShortTermMemoryTtlUpdate | Update policy for short-term memory TTL. Only
applies when `short-term-memory.state-ttl.ms` is greater than 0. Valid values:
`ON_CREATE_AND_WRITE`, `ON_READ_AND_WRITE`. An enabled run-begin memory
snapshot also refreshes TTL for entries it reads under `ON_READ_AND_WRITE`. |
-| `short-term-memory.state-ttl.visibility` | `NEVER_RETURN_EXPIRED` |
ShortTermMemoryTtlVisibility | Visibility policy for expired short-term memory
state. Only applies when `short-term-memory.state-ttl.ms` is greater than 0.
Valid values: `NEVER_RETURN_EXPIRED`, `RETURN_EXPIRED_IF_NOT_CLEANED_UP`.
|
-
-### Trace Log Options
-
-Trace Log records Events, Action executions, and calls made within Actions.
See [Monitoring]({{< ref "docs/operations/monitoring#trace-log" >}}) for output
formats and the Trace Tree reader.
-
-| Key | Default | Type | Description |
-|-----|---------|------|-------------|
-| `trace-log.output-type` | `SLF4J` | LoggerType | Trace log output: `SLF4J`
writes JSON through a dedicated SLF4J logger, visible in Flink's Web UI Logs
tab; `FILE` writes per-subtask files. A non-empty `trace-log.base-dir` forces
`FILE`, even when `SLF4J` is selected. |
-| `trace-log.base-dir` | (none) | String | Directory for file output; a
non-empty value forces `FILE`. When file output is selected without a
directory, uses `java.io.tmpdir/flink-agents`. |
-| `trace-log.pretty-print` | false | boolean | Format JSON over multiple
lines. When false, each record occupies one line (JSONL). |
-| `trace-log.targets` | `[{scope: EVENT_ONLY, level: STANDARD}]` | List of
objects | Match records and control the recording level. Each target uses
`scope` and optional `level`; see [Trace log targets](#trace-log-targets). |
-| `trace-log.standard.max-string-length` | 2000 | int | Maximum attribute
string length at `STANDARD` level. No effect at `VERBOSE`. |
-| `trace-log.standard.max-array-elements` | 20 | int | Maximum attribute array
length at `STANDARD` level. No effect at `VERBOSE`. |
-| `trace-log.standard.max-depth` | 5 | int | Maximum attribute nesting depth
at `STANDARD` level. Deeper objects are summarized. No effect at `VERBOSE`. |
-
-#### Trace log targets
-
-Every target combines a `scope` with an optional `level`. Omitting `targets`
records only Events at `STANDARD`. An explicit list replaces that default; an
empty list records nothing.
-
-For example, this configuration records all entities at `STANDARD`, with full
attributes for the `search` Tool:
-
-```yaml
-agent:
- trace-log:
- targets:
- - scope: ALL
- level: STANDARD
- - scope: {tool: search}
- level: VERBOSE
- standard:
- max-string-length: 2000
- max-array-elements: 20
- max-depth: 5
- output-type: FILE
- base-dir: /tmp/flink-agent-logs
- pretty-print: false
-```
+<span id="core-options"></span>
-Java and Python can configure targets with `TraceLogTarget` through the same
`TRACE_LOG_TARGETS` option:
+### Observability
-{{< tabs >}}
-{{< tab "python" >}}
+See [Monitoring]({{< ref "docs/operations/monitoring" >}}) for Trace Log
output and
+record selection, and [Memory Observability]({{< ref
"docs/development/memory/memory_events" >}})
+for the agent-run begin event and custom Event listeners.
-```python
-from flink_agents.api.core_options import AgentConfigOptions
-from flink_agents.api.trace_log import TraceLogLevel, TraceLogTarget
-
-agents_env.get_config().set(
- AgentConfigOptions.TRACE_LOG_TARGETS,
- [
- TraceLogTarget.all(TraceLogLevel.STANDARD),
- TraceLogTarget.tool("search", TraceLogLevel.VERBOSE),
- ],
-)
-```
+<span id="trace-log-options"></span>
-{{< /tab >}}
-{{< tab "java" >}}
-
-```java
-import java.util.List;
-import org.apache.flink.agents.api.configuration.AgentConfigOptions;
-import org.apache.flink.agents.api.logger.TraceLogLevel;
-import org.apache.flink.agents.api.logger.TraceLogTarget;
-
-agentsEnv.getConfig().set(
- AgentConfigOptions.TRACE_LOG_TARGETS,
- List.of(
- TraceLogTarget.all(TraceLogLevel.STANDARD),
- TraceLogTarget.tool("search", TraceLogLevel.VERBOSE)));
-```
-
-{{< /tab >}}
-{{< /tabs >}}
+| Key | Default | Type | Description |
+| --- | --- | --- | --- |
+| `agent-run.begin-event` | false | boolean | Emit an `AgentRunBeginEvent`
with a short-term memory value snapshot before running the Actions triggered by
each input. Independent of `memory.generate-event`. |
+| `event-listeners` | none | `List<String>` | Fully qualified Java
`EventListener` class names. Each class must provide a public no-argument
constructor. Python listener implementations are not supported. |
+| `trace-log.base-dir` | (none) | String | Directory for Trace Log files. A
nonblank value selects `FILE`, overriding `trace-log.output-type`. When unset,
file output uses `${java.io.tmpdir}/flink-agents`. |
+| `trace-log.output-type` | `SLF4J` | LoggerType | Trace Log destination:
`SLF4J` uses the Trace Log logger; `FILE` writes one file per operator subtask.
See [Output]({{< ref "docs/operations/monitoring#output" >}}) for logging
configuration and file locations. |
+| `trace-log.pretty-print` | false | boolean | Write multiline JSON when
`true`, or compact single-line JSON when `false`. Applies to both output types.
|
+| `trace-log.standard.max-array-elements` | 20 | int | Maximum elements
retained in each attribute array at `STANDARD` level. `0` disables this limit.
No effect at `VERBOSE`. |
+| `trace-log.standard.max-depth` | 5 | int | Nesting depth at which attribute
objects are summarized at `STANDARD` level. `0` disables this limit. No effect
at `VERBOSE`. |
+| `trace-log.standard.max-string-length` | 2000 | int | Maximum characters
retained in each attribute string at `STANDARD` level. `0` disables this limit.
No effect at `VERBOSE`. |
+| `trace-log.targets` | `[{scope: EVENT_ONLY, level: STANDARD}]` | List of
objects | Select Trace Log entities and their recording levels. Each target
requires `scope` and may specify `level`; see [Trace log
targets](#trace-log-targets). |
-`TraceLogTarget` provides `all()` and whole-type factories `events()`,
`actions()`, `tools()`, `llms()`, `parsers()`, and `subagents()`. The
corresponding singular factories accept a name, such as `event("OrderCreated")`
or `tool("search")`. All factories accept an optional `TraceLogLevel`. Omitting
it preserves inheritance: `tool("search")` inherits the level of a matching
Tool type target or `ALL`, as described below. Targets are immutable.
+#### Trace log targets
-In YAML, `scope` accepts `EVENT_ONLY` (match Events), `ALL` (match all
entities), or a selector object. Use one of the keys `event`, `action`, `tool`,
`llm`, `parser`, or `subagent` with an entity name, for example `{event:
OrderCreated}`. The value `"*"` selects that whole type: `{event: "*"}` is
equivalent to `EVENT_ONLY`.
+`trace-log.targets` is a list of target objects. Omitting the option records
Events
+at `STANDARD`. An explicit list replaces this default; an empty list records
nothing.
-The generic form `{entityType: custom, entityName: name}` also supports custom
entity types reported through `ExecutionReporter`; omit `entityName` to select
the whole type. Use `TraceLogTarget.entity(...)` for the same selection in Java
or Python. For example, `TraceLogTarget.entity("retriever", "catalog")`
corresponds to `scope: {entityType: retriever, entityName: catalog}`. Each
selector object must contain exactly one category key, or the generic
`entityType` and optional `entityName [...]
+Each target has the following fields:
-Direct model or tool calls inside an Action are not automatically recorded as
separate calls, even with `scope: ALL`.
+- `scope` (required): `EVENT_ONLY`, `ALL`, a category selector with exactly
one of
+ `event`, `action`, `tool`, `llm`, `parser`, or `subagent`, or a generic
selector with
+ `entityType` and optional `entityName`. Do not mix category and generic
fields.
+ A category value of `"*"`, or a generic selector without `entityName`,
selects the
+ whole entity type. `EVENT_ONLY` is equivalent to `{event: "*"}`.
+- `level` (optional): `OFF`, `STANDARD`, or `VERBOSE`. An omitted level
inherits
+ from matching scopes; it does not automatically mean `STANDARD`. In YAML,
quote
+ `level: "OFF"` so Python does not interpret it as a boolean.
-`entityType` identifies the kind of object to observe. For `entityName`, use
the logical name from your Agent's code or configuration. Trace Log records and
target selectors use the same name. Common examples are:
+Duplicate scopes may combine an omitted level with an explicit level or repeat
the
+same explicit level. Conflicting explicit levels for the same scope are
rejected
+when logging starts. This also applies to equivalent Event type selectors.
-| `entityType` | What to use as `entityName` | Example |
-|--------------|----------------------------|---------|
-| `event` | The Event type used when emitting or subscribing to Events. |
`OrderCreated` |
-| `action` | The name used to declare the Action in the Agent. |
`create_order` |
-| `llm` | The registered ChatModel Resource name used by the call. |
`primary_model` |
+For YAML and API examples, entity naming, and level inheritance, see
+[Recording Targets]({{< ref "docs/operations/monitoring#recording-targets"
>}}).
-For example, `scope: {llm: primary_model}` selects calls to the Resource named
`primary_model`. Its configured model or deployment is recorded separately in
`entityMetadata.model`.
+The old `event-log.*`, `eventLoggerType`, `baseLogDir`, and `prettyPrint` keys
are
+unsupported and rejected at startup. Use `trace-log.targets`,
`trace-log.standard.*`,
+and the output options above.
-Entity names match exactly. A name ending in `.*` matches its namespace and
descendants: `{event: "com.foo.*"}` matches `com.foo` and `com.foo.Event`, but
not `com.foobar.Event`. Other wildcard forms are unsupported. A category
selector with `"*"`, or a generic selector without `entityName`, matches the
whole type.
+### Condition handling
-`level` accepts `OFF`, `STANDARD`, and `VERBOSE`. `OFF` writes no record.
`STANDARD` truncates or summarizes attributes using the `standard` limits;
`VERBOSE` retains full attributes. Both recording settings preserve all
identity and relationship fields, entity metadata, timestamps, statuses, and
problem categories. Media payload sanitization applies at both recording
settings. See [Recording Level]({{< ref
"docs/operations/monitoring#recording-level" >}}) for the truncation format and
m [...]
+See [Condition Evaluation Failures]({{< ref
"docs/development/create-agent/actions#condition-evaluation-failures" >}})
+for trigger condition examples.
-For each entity, matching scopes are considered from most to least specific:
+| Key | Default | Type | Description |
+| --- | --- | --- | --- |
+| `action.trigger-condition.evaluate-failure-strategy` | `WARN_AND_SKIP` |
ConditionEvaluationFailureStrategy | Behavior when evaluating a trigger
condition for an Event fails. `WARN_AND_SKIP` logs a warning and treats the
condition as false; `FAIL` fails the Flink task. Initialization failures are
not affected. See [Condition Evaluation Failures]({{< ref
"docs/development/create-agent/actions#condition-evaluation-failures" >}}). |
-| Order | Matching scope |
-|-------|----------------|
-| 1 | Exact entity name within its type |
-| 2 | Namespace prefixes within its type, from longest to shortest |
-| 3 | Entity type (`EVENT_ONLY` is the `event` type scope) |
-| 4 | `ALL` |
+### Model calls, tools, and retries
-The first explicitly specified `level` in this chain applies. An omitted
`level` inherits from the next matching scope and does not override it. If at
least one scope matches but no matching scope specifies `level`, the entity
uses `STANDARD`. If no scope matches, no record is written. Target order does
not affect the result.
+See [Chat]({{< ref
"docs/development/calling-llms/chat#handle-success-and-failure" >}})
+and [Decision Models]({{< ref
"docs/development/decision_models#event-driven-calls" >}})
+for model-call outcomes and retries, and [Context Retrieval]({{< ref
"docs/development/vector-search/vector_stores#built-in-events-and-actions" >}})
+for vector-store queries. See [Tool Use]({{< ref
"docs/development/calling-llms/tool_use#parallel-tool-call-batches" >}})
+for tool-call batches and timeout behavior, and [Execution Metrics]({{< ref
"docs/operations/monitoring#execution-metrics" >}})
+for retry metrics.
-An explicit `OFF` stops inheritance and suppresses recording. A more specific
scope can enable recording by explicitly setting `STANDARD` or `VERBOSE`;
omitting `level` continues to inherit `OFF`.
+| Key | Default | Type | Description |
+| --- | --- | --- | --- |
+| `chat.async` | true | boolean | Enable durable async execution for built-in
Chat model calls, including routing judge calls. `false` selects synchronous
execution. |
+| `decision.async` | true | boolean | Enable durable async execution for
built-in Decision Model calls. `false` selects synchronous execution. |
+| `max-retries` | 0 | int | Maximum retries after the initial attempt of a
built-in Chat or Decision Model call, including routing judge calls.
Non-positive values disable retries. |
+| `rag.async` | true | boolean | Enable durable async execution for
vector-store queries made by the built-in context retrieval Action. `false`
selects synchronous execution. |
+| `retry-wait-interval` | 1 | int | Base wait in seconds between retries
controlled by `max-retries`. The wait before retry N is `retry-wait-interval *
2^(N-1)`. Non-positive values retry without waiting. |
+| `tool-call.async` | true | boolean | Enable durable async execution for
ordinary tools handled by the built-in Tool Call Action. With `false`, these
calls run synchronously, one at a time. See [Tool Use]({{< ref
"docs/development/calling-llms/tool_use#parallel-tool-call-batches" >}}) for
batch behavior. |
+
+These async switches select how built-in calls execute. Synchronous calls still
+use durable execution; saving outcomes for recovery requires an
+[Action State Store]({{< ref
"docs/operations/consistency#configure-an-action-state-store" >}}).
+Built-in Chat and context retrieval calls, and Java Decision Model calls
through
+Python, fall back to synchronous execution when the runtime cannot support
their
+cross-language async call. See [Execution Requirements]({{< ref
"docs/development/create-agent/actions#execution-requirements" >}})
+for cross-language async requirements.
+
+The `ErrorHandlingStrategy` API and `error-handling-strategy` option have been
+removed. Configure retries with `max-retries`; see Chat for terminal failure
handling.
+
+### Runtime and async execution
+
+See [Execution Requirements]({{< ref
"docs/development/create-agent/actions#execution-requirements" >}})
+for JDK, scheduling, and cross-language requirements. Batch limits also apply
to
+the built-in [parallel tool-call batches]({{< ref
"docs/development/calling-llms/tool_use#parallel-tool-call-batches" >}}).
-For example, a Tool type target at `VERBOSE` supplies the level to matching
namespace and exact-name targets that omit it. A matching namespace target at
`OFF` suppresses an exact-name target that omits `level`, while an exact-name
target with explicit `STANDARD` or `VERBOSE` enables that entity. `EVENT_ONLY`
applies only to Events; its level does not affect a Tool target.
+| Key | Default | Type | Description |
+| --- | --- | --- | --- |
+| `async.batch.parallelism` | os cpu count | int | Maximum concurrent calls in
one `ctx.gather(...)` batch, including ordinary, durable, and built-in tool
calls. Values of `1` or less run calls sequentially. Worker-pool capacity can
further limit concurrency. |
+| `async.batch.timeout.ms` | -1 (disabled) | long (milliseconds) | Overall
timeout for one `ctx.gather(...)` batch, in milliseconds. Non-positive values
disable it. A timeout does not guarantee that running calls stop and cannot
undo their external effects. See [Tool Use]({{< ref
"docs/development/calling-llms/tool_use#timeouts-and-external-effects" >}}) for
built-in tool-batch behavior. |
+| `job-identifier` | none | String | Stable job identifier used to scope Mem0
long-term memory. Keep an explicit value unchanged across restores. If unset,
the initial Flink JobID is stored in operator state and reused when restoring a
checkpoint or savepoint. |
+| `max-in-flight-input-records` | 100 | int | Maximum input records in flight
per Agent operator subtask on the experimental parallel engine. Must be
positive. The limit is ignored by continuation-based execution and plans
containing Python Actions. See [Execution Requirements]({{< ref
"docs/development/create-agent/actions#execution-requirements" >}}). |
+| `num-async-threads` | os cpu count * 2 | int | Worker threads per async
execution pool on each Agent operator subtask. Must be positive.
`async.batch.parallelism` separately limits concurrency within each batch; see
[Execution Requirements]({{< ref
"docs/development/create-agent/actions#execution-requirements" >}}) for
scheduling behavior. |
+| `parallel-execution.enabled` | true | boolean | **Experimental.** Enable the
parallel engine for eligible plans with only Java Actions when
continuation-based execution is unavailable. `false` selects serial execution.
Ignored by the continuation engine and plans containing Python Actions. See
[Execution Requirements]({{< ref
"docs/development/create-agent/actions#execution-requirements" >}}) for JDK and
Python Resource requirements. |
-Duplicate targets for the same scope merge an omitted level with an explicit
level and accept repeated identical explicit levels. Different explicit levels
for the same scope are rejected when logging starts, regardless of target
order. This includes equivalent targets expressed as `EVENT_ONLY`, `{event:
"*"}`, or `{entityType: event}`. For example, `EVENT_ONLY` without `level` and
`{event: "*"}` with `VERBOSE` merge into one Event type target at `VERBOSE`.
Typed targets, mapping objects [...]
+### Memory
-Selecting a Tool does not automatically select its parent Action; recorded
relationships retain their IDs even when related records are omitted.
+#### Memory events {#memory-event-options}
-An `ALL` preset with `level: "OFF"` disables its default range, while a more
specific target with explicit `STANDARD` or `VERBOSE` can enable recording. A
preset is unnecessary when selecting only one entity type. To record only Tools:
+An explicitly configured operation-specific switch overrides
`memory.generate-event`.
+If neither is set, the effective default below applies. See
+[Memory Observability]({{< ref "docs/development/memory/memory_events" >}})
for event
+payloads and subscription examples. The independent `agent-run.begin-event`
switch
+is listed under [Observability](#observability).
-```yaml
-agent:
- trace-log:
- targets:
- - scope: {tool: "*"}
- level: VERBOSE
-```
+| Key | Raw default | Effective default | Type | Description |
+| --- | --- | --- | --- | --- |
+| `memory.generate-event` | unset | per-operation defaults | boolean |
Fallback value for operation-specific memory-event switches that are unset. |
+| `memory.generate-event.long-term-get` | unset | on | boolean | Emit Events
describing long-term memory reads. |
+| `memory.generate-event.long-term-search` | unset | on | boolean | Emit
Events describing long-term memory searches. |
+| `memory.generate-event.long-term-update` | unset | on | boolean | Emit
Events describing long-term memory additions, updates, deletions, and
memory-set deletion. |
+| `memory.generate-event.sensory-read` | unset | off | boolean | Emit Events
describing sensory memory reads. |
+| `memory.generate-event.sensory-write` | unset | on | boolean | Emit Events
describing sensory memory writes. |
+| `memory.generate-event.short-term-read` | unset | off | boolean | Emit
Events describing short-term memory reads. |
+| `memory.generate-event.short-term-write` | unset | on | boolean | Emit
Events describing short-term memory writes. |
-To record everything except one Tool, use a local `OFF` target:
+#### Short-term memory TTL
-```yaml
-agent:
- trace-log:
- targets:
- - scope: ALL
- level: STANDARD
- - scope: {tool: search}
- level: "OFF"
-```
+See [Short-Term Memory TTL]({{< ref
"docs/development/memory/sensory_and_short_term_memory#short-term-memory-2" >}})
+for TTL behavior and setup examples.
-The old `event-log.*`, `eventLoggerType`, `baseLogDir`, and `prettyPrint`
configuration keys are unsupported and rejected at startup. Migrate selection
and verbosity to `targets`, truncation limits to `trace-log.standard.*`, and
output settings to `trace-log.output-type`, `trace-log.base-dir`, and
`trace-log.pretty-print`.
+| Key | Default | Type | Description |
+| --- | --- | --- | --- |
+| `short-term-memory.state-ttl.ms` | 0 | long | Short-term memory TTL in
milliseconds. A positive value enables expiration; non-positive values disable
it. |
+| `short-term-memory.state-ttl.update-type` | `ON_READ_AND_WRITE` |
ShortTermMemoryTtlUpdate | When TTL is enabled, refresh expiration on creation
and writes (`ON_CREATE_AND_WRITE`), or also on reads (`ON_READ_AND_WRITE`). |
+| `short-term-memory.state-ttl.visibility` | `NEVER_RETURN_EXPIRED` |
ShortTermMemoryTtlVisibility | When TTL is enabled, never return expired state
(`NEVER_RETURN_EXPIRED`), or allow it until cleanup
(`RETURN_EXPIRED_IF_NOT_CLEANED_UP`). |
-### Memory Event Options
+### Action consistency
-The eight `memory.generate-event*` options have no raw `ConfigOption` default.
When a sub-key and the master switch are both unset, the runtime uses the
effective default shown below. See [Memory Observability]({{< ref
"docs/development/memory/memory_events" >}}) for resolution order, event
payloads, and subscription examples.
+<span id="action-state-store"></span>
-| Key | Raw default | Effective default | Type | Description |
-|-----|-------------|-------------------|------|-------------|
-| `memory.generate-event` | unset | per-operation defaults | boolean | Master
fallback for unset operation-specific switches. |
-| `memory.generate-event.short-term-write` | unset | on | boolean | Emit
short-term memory write events. |
-| `memory.generate-event.short-term-read` | unset | off | boolean | Emit
short-term memory read events. |
-| `memory.generate-event.sensory-write` | unset | on | boolean | Emit sensory
memory write events. |
-| `memory.generate-event.sensory-read` | unset | off | boolean | Emit sensory
memory read events. |
-| `memory.generate-event.long-term-update` | unset | on | boolean | Emit
long-term memory add/delete events. |
-| `memory.generate-event.long-term-get` | unset | on | boolean | Emit
long-term memory get events. |
-| `memory.generate-event.long-term-search` | unset | on | boolean | Emit
long-term memory search events. |
-| `agent-run.begin-event` | false | off | boolean | Opt in to the agent-run
begin event. Independent of the memory-event master switch. |
-
-### Action State Store
+For consistency modes, setup examples, and recovery behavior, see
+[Consistency]({{< ref "docs/operations/consistency" >}}).
#### Common
| Key | Default | Type | Description
|
|------------------------------|------------------|---------|------------------------------------------------------------------------------------------|
-| `actionStateStoreBackend` | (none) | String | The backend for
action state store. Supported values: `"kafka"`, `"fluss"`. |
+| `actionStateStoreBackend` | (none) | String | Action State Store backend:
`"kafka"` or `"fluss"` (case-insensitive). When unset, no external Action State
Store is enabled. |
#### Kafka-based Action State Store
-Here are the configuration options for Kafka-based Action State Store.
+Dedicate one topic to each logical Agent operator, shared by that operator's
+subtasks. Separate operators in the same job need separate topics.
-| Key | Default | Type |
Description |
-|-------------------------------------|--------------------------|---------|-----------------------------------------------------------------------------|
-| `kafkaBootstrapServers` | "localhost:9092" | String |
The config parameter specifies the Kafka bootstrap server. |
-| `kafkaActionStateTopic` | (none) | String |
The Kafka topic for action state. Dedicate it to one logical Flink Agents
operator, shared by that operator's subtasks. |
-| `kafkaActionStateTopicNumPartitions`| 64 | Integer |
The config parameter specifies the number of partitions for the Kafka action
state topic. |
-| `kafkaActionStateTopicReplicationFactor` | 1 | Integer |
The config parameter specifies the replication factor for the Kafka action
state topic. |
-| `kafkaActionStateTombstoneEnabled` | false | Boolean |
Whether pruning sends tombstone records so log compaction can reclaim pruned
keys on a compacted action-state topic. Off by default: pruning does not
invalidate older restore points, but the topic continues to grow. When enabled,
the checkpoint whose completion triggers pruning remains usable, but restoring
an earlier checkpoint or savepoint may replay later tombstones and re-execute
already completed actions. E [...]
-| `kafkaActionStateCleanupControlTopic` | (none) | String |
Separate, single-partition topic containing committed checkpoint-aligned
cleanup boundaries. It must use `cleanup.policy=compact` without delete
retention, differ from the action-state topic, and be dedicated to the same job
recovery history. Setting it enables boundary enforcement during recovery. It
cannot be combined with `kafkaActionStateTombstoneEnabled`. |
+The store creates the configured topic if it does not exist, and reuses an
existing
+topic without changing its settings. The partition-count and replication-factor
+options apply only when creating a topic.
-##### Checkpoint-aligned Kafka cleanup
-
-Checkpoint-aligned cleanup is an explicit administrative operation.
`KafkaActionStateCleanupTool` provides the same command-line workflow for Java
and Python jobs. Run it with the Flink Agents runtime JAR and the matching
Flink State Processor API JAR on the classpath:
-
-```text
-plan --checkpoint PATH (--operator-uid UID | --operator-uid-hash HASH)
--output FILE
-apply --plan FILE --bootstrap-servers SERVERS --control-topic TOPIC
[--replication-factor N]
-```
+The Kafka Action State Store and its cleanup tool currently expose no
configuration
+for passing Kafka client SASL/SSL settings.
-The `plan` command reads all recovery markers from the selected checkpoint or
savepoint through Flink's State Processor API, creates a deterministic plan
using the earliest required offset per partition, and refuses to overwrite an
existing output file. Legacy map markers can still be restored by a job, but
cannot authorize deletion because they do not identify the physical Kafka
topic. The `apply` command verifies the content-derived plan ID before
contacting Kafka. Plan files must cont [...]
-
-Review and retain the plan's deterministic JSON before applying it. The
coordinator verifies that the selected offsets are still available, writes
`COMMITTED` before calling Kafka `deleteRecords`, verifies every resulting
beginning offset, and then writes `APPLIED`. Reapplying the same plan retries
an interrupted committed operation without changing its boundary. Set
`kafkaActionStateCleanupControlTopic` on the job only after the first plan is
committed; recovery requires the configured [...]
-
-Physical prefix cleanup requires the action-state data topic to use
`cleanup.policy=compact,delete`, `retention.ms=-1`, and `retention.bytes=-1`.
The store uses those settings when it creates a new topic. Before applying
cleanup to an existing topic, alter it to those values. The `delete` policy
enables Kafka's `deleteRecords` API, while both retention limits remain
disabled so Kafka cannot independently retire records that a supported
checkpoint still needs. Apply validates these effect [...]
-
-For the first cleanup on an existing job, stop the job, create and apply the
plan from the recovery point that will become the oldest supported one, and
restart from that same point with `kafkaActionStateCleanupControlTopic`
configured. This avoids a failover window in which the old running job does not
yet know the committed boundary. Once the job is running with boundary
enforcement enabled, later forward-only plans may be applied while it runs. Run
only one `apply` operation at a time [...]
+| Key | Default | Type |
Description |
+|-------------------------------------|--------------------------|---------|-----------------------------------------------------------------------------|
+| `kafkaActionStateCleanupControlTopic` | (none) | String | Separate topic
containing committed checkpoint-aligned cleanup boundaries. When configured,
recovery rejects points requiring data below the boundary. Mutually exclusive
with `kafkaActionStateTombstoneEnabled=true`. See [Checkpoint-aligned Kafka
cleanup]({{< ref "docs/operations/consistency#checkpoint-aligned-kafka-cleanup"
>}}) for prerequisites and procedure. |
+| `kafkaActionStateTombstoneEnabled` | false | Boolean | Write deletion
markers for execution records covered by completed checkpoints. Enabling this
may cause completed Actions to run again when restoring older checkpoints or
savepoints. Mutually exclusive with `kafkaActionStateCleanupControlTopic`. See
[Kafka retention and cleanup]({{< ref
"docs/operations/consistency#kafka-retention-and-cleanup" >}}) for the recovery
trade-off. |
+| `kafkaActionStateTopic` | (none) | String | Topic storing Action state.
Required when `actionStateStoreBackend` is `kafka`. |
+| `kafkaActionStateTopicNumPartitions` | 64 | Integer | Number of partitions
when creating the topic. Does not change an existing topic. |
+| `kafkaActionStateTopicReplicationFactor` | 1 | Integer | Replication factor
when creating the topic. Does not change an existing topic. |
+| `kafkaBootstrapServers` | "localhost:9092" | String | Bootstrap servers for
the Kafka Action State Store. |
-The action-state topic and control topic must be dedicated to one job's
recovery history. Once any boundary has been committed, every future run and
restore of that history must retain the same control-topic configuration;
omitting or changing it removes logical boundary enforcement. The boundary can
move only forward. A checkpoint whose marker is below the committed boundary is
rejected even if Kafka has not finished physical deletion.
+##### Changes to an existing Kafka store
-Per-key tombstones and checkpoint-aligned cleanup are mutually exclusive.
Tombstones are not tied to the selected recovery boundary and could invalidate
a checkpoint that checkpoint-aligned cleanup promises to retain.
+When reusing checkpoints or savepoints with this store:
-To migrate a job that previously emitted tombstones, first stop it cleanly so
its Kafka producer closes and all accepted tombstone sends finish. Restart from
the newest recovery point that is valid under the tombstone mode's existing
recovery trade-off, with `kafkaActionStateTombstoneEnabled=false` and no
cleanup control topic configured. After that attempt completes a new checkpoint
or savepoint, stop it again, create and apply the first cleanup plan from that
new recovery point, and re [...]
+- Keep the original topic and partition count. Increasing the partition count
+ prevents recovery from existing snapshots and causes checkpoints of the
running
+ job to fail. Do not delete and recreate the topic or switch to a replacement,
+ even if it has the same name and partition count.
+- Retention changes and manual deletion must preserve the records needed by
every
+ recovery point you intend to retain. See
+ [Kafka retention and cleanup]({{< ref "docs/operations/consistency"
>}}#kafka-retention-and-cleanup).
#### Fluss-based Action State Store
-Here are the configuration options for Fluss-based Action State Store.
-
-When taking a checkpoint, the store captures the bucket end offsets and then
synchronously appends a fresh copy of each cached action state needed for
replay before saving that recovery marker. States whose input sequence is at or
below the key’s last completed sequence number in this checkpoint are skipped.
Completed actions in unfinished inputs and all states for keys without a
completed sequence number are still refreshed. This keeps durable call results
available even when they were [...]
+Dedicate one table to each logical Agent operator, shared by that operator's
+subtasks. Separate operators in the same job need separate tables.
-Fluss log retention must preserve the records from every checkpoint or
savepoint you intend to restore. Refreshing state does not protect older
restore points from retention or repair recovery markers created before this
fix.
+The store creates the configured database and append-only log table if they do
not
+exist, and reuses an existing table. The bucket-count option applies only when
+creating a table. SASL options apply when `flussSecurityProtocol` is `SASL`;
they
+are ignored for `PLAINTEXT`.
| Key | Default | Type | Description
|
|------------------------------|------------------|---------|------------------------------------------------------------------------------------------|
-| `flussBootstrapServers` | "localhost:9123" | String | The Fluss
bootstrap servers address. |
-| `flussActionStateDatabase` | "flink_agents" | String | The Fluss
database name for storing action state. |
-| `flussActionStateTable` | (none) | String | The Fluss table
for action state. Dedicate it to one logical Flink Agents operator, shared by
that operator's subtasks. |
-| `flussActionStateTableBuckets` | 64 | Integer | The number of
buckets for the Fluss action state table. |
-| `flussSecurityProtocol` | "PLAINTEXT" | String | The
authentication protocol for Fluss client. Valid values: `PLAINTEXT` (default,
no authentication), `SASL` (SASL/PLAIN authentication). |
-| `flussSaslMechanism` | "PLAIN" | String | The SASL
mechanism for Fluss authentication.
|
-| `flussSaslJaasConfig` | (none) | String | The JAAS
configuration string for Fluss SASL authentication.
|
-| `flussSaslUsername` | (none) | String | The username for
Fluss SASL authentication. |
-| `flussSaslPassword` | (none) | String | The password for
Fluss SASL authentication. |
+| `flussActionStateDatabase` | "flink_agents" | String | Database containing
the Action State Store table. |
+| `flussActionStateTable` | (none) | String | Append-only log table storing
Action state. Required when `actionStateStoreBackend` is `fluss`. |
+| `flussActionStateTableBuckets` | 64 | Integer | Number of buckets when
creating the table. Does not change an existing table. |
+| `flussBootstrapServers` | "localhost:9123" | String | Bootstrap servers for
the Fluss Action State Store. |
+| `flussSaslJaasConfig` | (none) | String | JAAS configuration for Fluss SASL
authentication. Takes precedence over `flussSaslUsername` and
`flussSaslPassword`. |
+| `flussSaslMechanism` | "PLAIN" | String | SASL mechanism used when
`flussSecurityProtocol` is `SASL`. Use `PLAIN` for SASL/PLAIN authentication. |
+| `flussSaslPassword` | (none) | String | Password for SASL/PLAIN
authentication. Set together with `flussSaslUsername` when
`flussSaslJaasConfig` is unset. |
+| `flussSaslUsername` | (none) | String | Username for SASL/PLAIN
authentication. Set together with `flussSaslPassword` when
`flussSaslJaasConfig` is unset. |
+| `flussSecurityProtocol` | "PLAINTEXT" | String | Fluss client authentication
protocol: `PLAINTEXT` (no authentication) or `SASL` (SASL/PLAIN
authentication), case-insensitive. |
+
+##### Changes to an existing Fluss store
+
+When reusing checkpoints or savepoints with this store:
+
+- Keep the original cluster, database, table, and bucket layout. Do not change
the
+ bucket count, recreate the table, or switch to a replacement table. A
replacement
+ with the same name and bucket count may not be detected during recovery.
+- Preserve the append-only log table type and the schema used by Flink Agents,
+ including column types and order.
+- Log retention changes and manual truncation must preserve the records needed
+ by every recovery point you intend to retain. See
+ [Fluss retention]({{< ref "docs/operations/consistency" >}}#fluss-retention).
diff --git a/docs/content/docs/operations/consistency.md
b/docs/content/docs/operations/consistency.md
new file mode 100644
index 000000000..07a6a312a
--- /dev/null
+++ b/docs/content/docs/operations/consistency.md
@@ -0,0 +1,356 @@
+---
+title: Consistency
+weight: 2
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+When a Flink job recovers from a checkpoint, it restores its state and replays
the
+input received since that checkpoint. An Agent may therefore call a model,
invoke
+a tool, or update an external system again. The consistency mode you choose
+determines what state must remain available for recovery and which work can be
+reused.
+
+## Choose a consistency mode
+
+With exactly-once checkpointing and the required source and sink guarantees,
+each input affects the job's state and output once, even if recovery processes
it
+again. Flink Agents calls this **exactly-once output consistency**.
+
+Checkpoint recovery may still rerun Actions and external calls, so you can add
an
+external Action State Store to reuse saved Action completions and durable call
+outcomes. Flink Agents calls this **exactly-once action consistency**.
+
+| Configuration | Behavior after a failure | Requirements and trade-offs |
+| --- | --- | --- |
+| No checkpointing | The job restarts from the source's configured starting
positions without restoring Agent state or progress from a checkpoint. |
Suitable when the application can accept lost progress or repeated work. |
+| Flink checkpointing | Flink restores checkpointed state and replays
subsequent input. Actions and external calls may repeat, and models or queries
may return different results. | Enable exactly-once checkpointing with durable
checkpoint storage and the required source and sink guarantees. External calls
must tolerate repetition. |
+| Flink checkpointing + Action State Store | Recovery replays saved
completions without rerunning Action bodies. Unfinished Actions may run again,
reusing saved durable call outcomes. | Add a Kafka or Fluss Action State Store.
Saving Action completions and durable call outcomes requires synchronous
backend writes. Retain both checkpoints and the backend records needed for
recovery. |
+
+## Required setup
+
+Enable exactly-once checkpointing for the Flink or PyFlink job, and use source
+and sink connectors with the required
+[fault tolerance
guarantees](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/connectors/datastream/guarantees/).
+Follow Flink's [checkpointing
guide](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/datastream/fault-tolerance/checkpointing/)
+for configuration and checkpoint storage, and
+[checkpoint recovery
guide](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/ops/state/checkpoints/#resuming-from-a-retained-checkpoint)
+for restore procedures. Use the Flink documentation for the version you deploy.
+
+### Configure an Action State Store
+
+Before building the Agent pipeline, configure Kafka or Fluss as the Action
State
+Store backend. Each logical Agent operator needs its **own Kafka topic or Fluss
+table**, shared only by that operator's subtasks.
+
+The Python and Java examples below use `agents_env` and `agentsEnv`,
respectively,
+created as described in [Integrate with Flink]({{< ref
"docs/development/integrate_with_flink" >}}).
+The YAML examples show the same settings in the `agent:` section of the job's
+Flink configuration.
+
+#### Kafka
+
+Set the backend to `kafka` and provide the broker address and a dedicated
topic:
+
+{{< tabs "Kafka Action State Store" >}}
+{{< tab "Python" >}}
+```python
+from flink_agents.api.core_options import AgentConfigOptions
+
+config = agents_env.get_config()
+config.set(AgentConfigOptions.ACTION_STATE_STORE_BACKEND, "kafka")
+config.set(AgentConfigOptions.KAFKA_BOOTSTRAP_SERVERS, "kafka:9092")
+config.set(AgentConfigOptions.KAFKA_ACTION_STATE_TOPIC,
"orders-agent-action-state")
+```
+{{< /tab >}}
+{{< tab "Java" >}}
+```java
+import org.apache.flink.agents.api.configuration.AgentConfigOptions;
+import org.apache.flink.agents.api.configuration.Configuration;
+
+Configuration config = agentsEnv.getConfig();
+config.set(AgentConfigOptions.ACTION_STATE_STORE_BACKEND, "kafka");
+config.set(AgentConfigOptions.KAFKA_BOOTSTRAP_SERVERS, "kafka:9092");
+config.set(AgentConfigOptions.KAFKA_ACTION_STATE_TOPIC,
"orders-agent-action-state");
+```
+{{< /tab >}}
+{{< tab "YAML" >}}
+```yaml
+agent:
+ actionStateStoreBackend: kafka
+ kafkaBootstrapServers: "kafka:9092"
+ kafkaActionStateTopic: orders-agent-action-state
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+If the topic does not exist, the store creates it with 64 partitions and a
+replication factor of 1 by default. Otherwise, it reuses the existing topic
without
+changing its settings.
+For retention requirements and cleanup options, see
+[Kafka retention and cleanup](#kafka-retention-and-cleanup).
+
+#### Fluss
+
+Set the backend to `fluss` and provide the server address and a dedicated
table:
+
+{{< tabs "Fluss Action State Store" >}}
+{{< tab "Python" >}}
+```python
+from flink_agents.api.core_options import AgentConfigOptions
+
+config = agents_env.get_config()
+config.set(AgentConfigOptions.ACTION_STATE_STORE_BACKEND, "fluss")
+config.set(AgentConfigOptions.FLUSS_BOOTSTRAP_SERVERS, "fluss:9123")
+config.set(AgentConfigOptions.FLUSS_ACTION_STATE_DATABASE, "flink_agents")
+config.set(AgentConfigOptions.FLUSS_ACTION_STATE_TABLE,
"orders_agent_action_state")
+```
+{{< /tab >}}
+{{< tab "Java" >}}
+```java
+import org.apache.flink.agents.api.configuration.AgentConfigOptions;
+import org.apache.flink.agents.api.configuration.Configuration;
+
+Configuration config = agentsEnv.getConfig();
+config.set(AgentConfigOptions.ACTION_STATE_STORE_BACKEND, "fluss");
+config.set(AgentConfigOptions.FLUSS_BOOTSTRAP_SERVERS, "fluss:9123");
+config.set(AgentConfigOptions.FLUSS_ACTION_STATE_DATABASE, "flink_agents");
+config.set(AgentConfigOptions.FLUSS_ACTION_STATE_TABLE,
"orders_agent_action_state");
+```
+{{< /tab >}}
+{{< tab "YAML" >}}
+```yaml
+agent:
+ actionStateStoreBackend: fluss
+ flussBootstrapServers: "fluss:9123"
+ flussActionStateDatabase: flink_agents
+ flussActionStateTable: orders_agent_action_state
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+The store creates the database and append-only log table if needed. By default,
+it uses the `flink_agents` database and creates new tables with 64 buckets.
+
+The store does not set log retention when creating a table, so configure
retention
+to preserve the records required by your recovery points. If the cluster
requires
+authentication, configure the Fluss authentication options as well. See
+[Configuration]({{< ref "docs/operations/configuration" >}}) for these
settings,
+option defaults, and configuration loading and precedence.
+
+## Recovery guarantees and user responsibilities
+
+### Recovery flow
+
+When restoring a checkpoint or savepoint with its corresponding Action State
+Store, Flink Agents handles each Action as follows:
+
+1. **Check for a saved Action completion.** If one is available, Flink Agents
+ replays its saved Sensory/Short-Term Memory updates and output Events
without
+ running the Action body again. The replayed Events keep their original Event
+ IDs.
+2. **Rerun an unfinished Action.** If no completion is found, Flink Agents
runs the
+ Action body again. Ordinary calls may therefore repeat, but the framework
+ checks each durable call for a matching record before executing it:
+
+ - **Saved success or failure:** replay the saved outcome without calling the
+ function again.
+ - **Recorded in-flight call:** run its reconciler if one is configured.
+ Otherwise, execute the function again.
+ - **No matching record:** execute the function normally.
+
+### Conditions for reusing saved records
+
+To match saved records to the correct work, replay must preserve the following:
+
+- **Preserve input order for each key.** Recovery must receive each key's
inputs
+ in the same order as the original execution.
+- **For custom Actions, keep the durable call sequence stable.** An unfinished
+ Action must reach the same durable calls in the same order, with the same
inputs.
+ See [Durable Execution]({{< ref
"docs/development/create-agent/actions#durable-execution" >}})
+ for call identity, serialization, and Action-context requirements.
+- **Keep keys and Action names stable.** Reuse the original key types and
+ serializer configuration, and ensure custom key serializers produce
+ deterministic bytes. Changing key serialization or renaming an Action can
+ prevent recovery from finding its saved completion.
+- **Keep saved values usable.** The application must still be able to read and
use
+ stored Events, Memory updates, and durable call outcomes, since recovery may
+ replay them without running the code that produced them.
+
+### Limits on external side effects
+
+The Action State Store cannot commit an external operation and its saved
outcome
+as a single transaction. If the operation finishes but the job fails before its
+outcome is saved, recovery may execute it again.
+
+For operations that change external state, use the service's idempotency
support
+or, for a durable call, provide a reconciler that can recover the original
outcome.
+See [Reconciling an in-flight call]({{< ref
"docs/development/create-agent/actions#reconciling-an-in-flight-call" >}})
+for the API and requirements.
+
+## Action State Store retention and cleanup
+
+For every checkpoint or savepoint you may need to restore, retain both the
+snapshot and the Action State Store records it requires. Before changing
retention
+or cleanup settings, identify the oldest recovery point you need to keep.
+
+{{< hint warning >}}
+Hashing business keys does not encrypt Action state. Saved Events, Memory
updates,
+and durable call outcomes may contain sensitive data. Protect access to these
+records.
+{{< /hint >}}
+
+### Kafka retention and cleanup
+
+Topics created by the store use `cleanup.policy=compact,delete`,
+`retention.ms=-1`, and `retention.bytes=-1`. The `-1` values disable the time
and
+size limits for the `delete` policy. Compaction can still reclaim older values
+for the same key.
+
+For an existing topic, check that its retention settings preserve the records
+needed by every recovery point you intend to keep.
+
+A completed checkpoint includes the Agent's progress for the work it covers,
+which means those execution records are no longer needed to restore it. Because
+earlier recovery points may still need the same records, choose a cleanup
method
+based on the recovery history you need to retain:
+
+| Method | Cleanup behavior | When to use it | Recovery impact |
+| --- | --- | --- | --- |
+| No tombstones (default) | With `kafkaActionStateTombstoneEnabled=false`,
completed work is removed from the in-memory cache, while its Kafka records
remain. | Kafka storage growth is acceptable and you need to retain older
recovery points. | Cache cleanup alone does not invalidate older recovery
points. Kafka retention or manual deletion can still remove required data. |
+| Per-key tombstones | The store automatically writes deletion markers for
execution records covered by each completed checkpoint. Kafka can reclaim those
records through background compaction. | The job only needs the latest
completed checkpoint, or repeating affected Actions is acceptable. | The
triggering checkpoint remains usable. Restoring an earlier checkpoint or
savepoint may repeat completed Actions. |
+| Checkpoint-aligned cleanup | An administrator selects the oldest recovery
point to retain. The tool records a cleanup boundary in a separate control
topic, then deletes older Kafka data no longer needed by the selected recovery
point. | You need to retain a selected checkpoint or savepoint while reclaiming
space from older recovery data. | The selected point remains usable. With the
control topic configured, recovery points requiring data below the boundary are
rejected. |
+
+Per-key tombstones and checkpoint-aligned cleanup cannot be enabled together.
+A committed cleanup boundary takes effect even if physical deletion has not
+finished.
+
+#### Cleanup steps
+
+For per-key tombstones, set `kafkaActionStateTombstoneEnabled=true` before
starting
+the job.
+
+##### Checkpoint-aligned Kafka cleanup
+
+Use `KafkaActionStateCleanupTool` for both Java and Python jobs. Before
generating
+a plan, check the prerequisites below. If the job has used per-key tombstones,
+first follow [Switching from per-key
tombstones](#switching-from-per-key-tombstones).
+
+{{< details title="Prerequisites and topic settings" >}}
+
+- The data topic must use `cleanup.policy=compact,delete`, `retention.ms=-1`,
and
+ `retention.bytes=-1`. These are also the settings used when the store
creates a
+ new data topic.
+- The control topic must differ from the data topic, have exactly one
partition,
+ and use `cleanup.policy=compact` without delete retention. Dedicate it to the
+ same logical Agent operator and physical data topic; do not share it across
+ operators.
+- Before enabling `kafkaActionStateCleanupControlTopic` on the job, the control
+ topic must exist and contain a committed cleanup boundary. Recovery fails if
+ the topic or its committed history is missing.
+- The `plan` command accepts only versioned Kafka recovery markers. Legacy map
+ markers can still be used for job recovery, but cannot authorize cleanup.
+
+{{< /details >}}
+
+For the first cleanup of an existing job:
+
+1. Select the oldest completed checkpoint or savepoint to retain, and stop the
+ job while keeping that snapshot available.
+2. Generate a cleanup plan for the Agent operator in that snapshot, and check
+ that it matches the selected recovery point and Action State Store.
+3. Apply the plan with a separate cleanup control topic dedicated to that
Action
+ State Store.
+4. After `apply` succeeds, set `kafkaActionStateCleanupControlTopic` to that
topic
+ and `kafkaActionStateTombstoneEnabled=false`, then restart from the
+ **same selected recovery point**.
+
+{{< details title="Command examples" >}}
+
+Run the tool where the recovery point and Kafka are accessible. The classpath
+must include the full Flink Agents distribution JAR for your Flink version
+(`flink-agents-dist-flink-<FLINK_VERSION>-<FA_VERSION>.jar`, without the
`-thin`
+classifier), the matching Flink State Processor API JAR, and Flink's `lib/`.
+
+Generate a plan with the Agent operator's UID hash in the selected snapshot.
+Replace the placeholders below, or use `--operator-uid` instead of
+`--operator-uid-hash` when specifying a UID:
+
+```bash
+java -cp
"$FLINK_HOME/lib/*:<FLINK_AGENTS_DIST_JAR>:<FLINK_STATE_PROCESSOR_API_JAR>" \
+ org.apache.flink.agents.runtime.actionstate.KafkaActionStateCleanupTool \
+ plan --checkpoint "<CHECKPOINT_OR_SAVEPOINT_PATH>" \
+ --operator-uid-hash "<AGENT_OPERATOR_UID_HASH>" \
+ --output cleanup-plan.json
+```
+
+After checking the plan, apply it with a separate control topic. The tool
creates
+the topic if it does not exist.
+
+```bash
+java -cp
"$FLINK_HOME/lib/*:<FLINK_AGENTS_DIST_JAR>:<FLINK_STATE_PROCESSOR_API_JAR>" \
+ org.apache.flink.agents.runtime.actionstate.KafkaActionStateCleanupTool \
+ apply --plan cleanup-plan.json --bootstrap-servers "kafka:9092" \
+ --control-topic "orders-agent-action-state-cleanup"
+```
+
+To choose the replication factor when creating the control topic, add
+`--replication-factor N` to `apply`. The default is `1`; valid values are `1`
+through `32767`.
+
+A successful apply prints `Cleanup plan <id> is APPLIED`.
+
+{{< /details >}}
+
+If the first cleanup fails or is interrupted, keep the job stopped and retry
+`apply` with the same plan: the boundary may already have been committed,
making
+earlier recovery points unavailable.
+
+Once boundary enforcement is enabled, later plans may be applied while the job
+runs. Apply one plan at a time, only advance the boundary, and keep the data
topic
+and its partition count unchanged during cleanup.
+
+{{< hint warning >}}
+Keep the same cleanup control topic configured and retain its committed records
+for every subsequent run and restore of this recovery history. Omitting or
+changing the topic removes boundary enforcement; with it configured, the job
also
+rejects a fresh start without a recovery marker.
+{{< /hint >}}
+
+#### Switching from per-key tombstones
+
+To switch a job from per-key tombstones to checkpoint-aligned cleanup:
+
+1. Stop the job cleanly and wait for pending tombstone writes to finish.
+2. Restore the newest recovery point that is still valid under the tombstone
+ policy. Set `kafkaActionStateTombstoneEnabled=false` and leave the cleanup
+ control-topic option unset.
+3. Complete and retain a new checkpoint or savepoint, then stop the job again.
+4. Use that new recovery point for the first cleanup plan and apply operation.
+5. Restart from the same point with the cleanup control topic configured.
+
+Use this new recovery point because an older one may still require records
+affected by past tombstones.
+
+### Fluss retention
+
+Flink Agents currently provides no equivalent of Kafka's per-key tombstones or
+checkpoint-aligned cleanup for the Fluss Action State Store. Instead, use
Fluss's
+own log retention settings to manage record retention and cleanup.
diff --git a/docs/content/docs/operations/deployment.md
b/docs/content/docs/operations/deployment.md
index 8d9ee4e96..46aad7c3e 100644
--- a/docs/content/docs/operations/deployment.md
+++ b/docs/content/docs/operations/deployment.md
@@ -84,60 +84,11 @@ For all available options, see the [Flink CLI
documentation](https://nightlies.a
Now you should see a Flink job submitted to the Flink Cluster in Flink web UI
(typically accessible at http://<jobmanagerHost>:8081).
-## Consistency Guarantees
-
-Flink Agents leverage Flink’s checkpointing to provide **exactly-once output
consistency**. Exactly-once output consistency does not mean each event is
processed only once; it means that every event will affect the state and output
result exactly once. This is sufficient for typical data-processing use cases
where the processing of data does not cause any external side effects.
-
-However, actions in Flink Agents can trigger external side effects (for
example, sending an email or invoking an external API). These side effects
should not be executed more than once. To address this, Flink Agents support
**exactly-once action consistency** by leveraging an external action state
store.
-
-### Exactly-Once Output Consistency
-
-When a job runs on a Flink cluster, Flink periodically takes checkpoints. See
[Flink
Checkpointing](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/ops/state/checkpoints/)
for details.
-After recovery from a checkpoint, Flink Agents reprocess events that arrived
after that checkpoint. As a result, any actions triggered by those events may
be executed again.
-
-### Exactly-Once Action Consistency
-
-To enable exactly-once action consistency, configure an external action state
store.
-Flink Agents record state for each Action in this store. After recovering from
a
-checkpoint, the framework reuses Actions whose completion was persisted,
including
-their stored Sensory/Short-Term Memory updates and output Events. Unfinished
Actions
-may execute again.
-
-When a completed Action is reused during recovery, its stored output Events
keep their original Event IDs. The runtime records their relationships to the
current trigger Event in Trace Log `entityMetadata`.
-
-The same store also saves individual call outcomes for
-[Durable Execution]({{< ref
"docs/development/create-agent/actions#durable-execution" >}}).
-When an unfinished Action runs again, durable calls can reuse persisted
outcomes.
-An external operation can finish before its outcome is saved, so recovery may
repeat
-that operation. Use idempotent operations or reconciliation for external side
effects;
-see [Recovery Behavior]({{< ref
"docs/development/create-agent/actions#recovery-behavior" >}}).
-
-{{< hint info >}}
-**Note**: Currently, Kafka and Fluss are supported as the external action
state store.
-{{< /hint >}}
-
-{{< hint warning >}}
-The action-state key format changed in Flink Agents 0.4 and action-state
records written by earlier versions are unsupported. When upgrading, use a
fresh Kafka topic or Fluss table and start without an older checkpoint or
savepoint. Recovery validates the current fields, but cannot reliably
distinguish every older record from the current format.
+<span id="exactly-once-output-consistency"></span>
+<span id="exactly-once-action-consistency"></span>
-Action-state keys contain a digest of the serialized key. Changing key types
or serializer configuration when recovering existing state is unsupported, even
if Flink accepts the change for keyed state: the same key can serialize to
different bytes, causing recovery to miss completed actions and repeat their
side effects. These changes are not detected by the action-state store.
Preserve the original key types and serializer configuration for recovery.
Custom key serializers must produce [...]
-
-Action-state keys identify an action by its name. Keeping an action's name
across a code change lets recovery reuse its completed results; renaming an
action makes its recorded results unreachable, so it is executed again on the
next recovery.
-
-Dedicate each Kafka topic or Fluss table to one logical Flink Agents operator,
shared by that operator's subtasks. Action-state keys do not contain a job or
operator namespace, so sharing a backend between logical operators can allow
otherwise identical records to collide.
-{{< /hint >}}
-
-The business-key component stored in the backend is a SHA-256 digest rather
than the serialized key itself. This keeps record keys bounded and avoids
embedding raw key bytes, but it is not encryption; protect the action-state
backend with appropriate access controls.
-
-See [Action State Store Configuration]({{< ref
"docs/operations/configuration#action-state-store" >}}) for configuration
options.
-
-{{< hint warning >}}
-**Note**: Enabling Kafka action-state tombstones can invalidate checkpoints or
savepoints older than the prune and cause completed actions to execute again.
See [Action State Store Configuration]({{< ref
"docs/operations/configuration#action-state-store" >}}) for the recovery
trade-off.
-{{< /hint >}}
-
-{{< hint warning >}}
-**Note**: A committed checkpoint-aligned Kafka cleanup boundary permanently
makes older recovery points unsupported. Every subsequent run and restore must
use the same cleanup control topic so the runtime can reject those recovery
points before replay. See [Checkpoint-aligned Kafka cleanup]({{< ref
"docs/operations/configuration#checkpoint-aligned-kafka-cleanup" >}}).
-{{< /hint >}}
+## Consistency Guarantees
-{{< hint info >}}
-**Note**: Exactly-once action consistency is guaranteed only if, after
recovering from the same checkpoint, inputs for each key arrive in the same
order as before recovery. If this ordering requirement is not met, the system
falls back to exactly-once output consistency.
-{{< /hint >}}
+Before submitting the job, choose and configure its consistency mode. See
+[Consistency]({{< ref "docs/operations/consistency" >}}) for required setup,
+recovery guarantees, retention and cleanup.
diff --git a/docs/content/docs/operations/monitoring.md
b/docs/content/docs/operations/monitoring.md
index 3e67f447b..93b762bc7 100644
--- a/docs/content/docs/operations/monitoring.md
+++ b/docs/content/docs/operations/monitoring.md
@@ -1,6 +1,6 @@
---
title: Monitoring
-weight: 3
+weight: 4
type: docs
---
<!--
@@ -100,7 +100,7 @@ These metrics describe LLM, Tool, Skill, MCP Server and
sub-agent calls that the
| **Model Resource** |
action.\<action_name\>.model_resource.\<resource_name\>.numOfLlmCallsSucceeded
| Model invocations that returned successfully. [1] [6] | Counter |
| **Model Resource** |
action.\<action_name\>.model_resource.\<resource_name\>.numOfLlmCallsFailed |
Model invocations that failed. [1] [6] | Counter |
| **Model Resource** |
action.\<action_name\>.model_resource.\<resource_name\>.llmCallLatencyMs | Time
for one model invocation, excluding structured-output parsing and retry waits.
[1] | Histogram |
-| **Model Resource** |
action.\<action_name\>.model_resource.\<resource_name\>.retryCount | Additional
invocations made by framework retries (`max-retries > 0`); recorded only when a
retry occurs. See [retry-wait-interval]({{< ref
"docs/operations/configuration#core-options" >}}). | Counter |
+| **Model Resource** |
action.\<action_name\>.model_resource.\<resource_name\>.retryCount | Additional
invocations made by framework retries (`max-retries > 0`); recorded only when a
retry occurs. See [retry-wait-interval]({{< ref
"docs/operations/configuration#model-calls-tools-and-retries" >}}). | Counter |
| **Model Resource** |
action.\<action_name\>.model_resource.\<resource_name\>.retryWaitSec | Backoff
time accumulated by framework retries; recorded only when a retry occurs. |
Counter |
| **Tool** | action.\<action_name\>.tool.\<tool_name\>.numOfToolCallsSucceeded
| Successful calls to the Tool. [2] [6] | Counter |
| **Tool** | action.\<action_name\>.tool.\<tool_name\>.numOfToolCallsFailed |
Failed calls to the Tool. [2] [6] | Counter |
@@ -233,15 +233,15 @@ You can read the logs of a running job in the Flink Web
UI:
## Trace Log
-Trace Log records Events, Action executions, and component calls as structured
JSON. By default, it records only Events at `STANDARD` level. Configure
[`trace-log.targets`]({{< ref "docs/operations/configuration#trace-log-targets"
>}}) to select other entities or change their level.
+Trace Log records Events, Action executions, and component calls as structured
JSON. By default, it records only Events at `STANDARD` level. Configure
[recording targets](#recording-targets) to select other entities or change
their level.
Trace Log writes are best effort. A failed write does not fail Event
processing or the job; the first failure is logged at `WARN`, later ones at
`DEBUG`, and each one increments `traceLogWriteFailures`.
### Recording Targets
-A target's `scope` supports `EVENT_ONLY` (the default), which records only
Events, and `ALL`, which records all entity types the framework observes. The
example below uses `ALL`.
+Use `trace-log.targets` to choose which entities to record and at which level.
Omitting it records only Events at `STANDARD`. An explicit target list replaces
that default; an empty list records nothing.
-To record Events, Actions, and component calls at `STANDARD`, with full
attributes for the `search` Tool, configure:
+For example, this configuration records Events, Actions, and component calls
at `STANDARD`, with full attributes for the `search` Tool:
```yaml
agent:
@@ -251,17 +251,120 @@ agent:
level: STANDARD
- scope: {tool: search}
level: VERBOSE
+ standard:
+ max-string-length: 2000
+ max-array-elements: 20
+ max-depth: 5
output-type: FILE
base-dir: /tmp/flink-agent-logs
+ pretty-print: false
```
-This example uses file output so the records can be read locally. Each target
combines a `scope` with an optional `level`. The specific Tool target overrides
the `ALL` preset. Use `level: "OFF"` to suppress a matching entity. Selecting
an entity does not automatically select its parents or children.
+This example uses file output so the records can be read locally. The
[Recording Level](#recording-level) section explains how the selected level and
the `standard` limits affect the recorded data.
-See [Trace log targets]({{< ref
"docs/operations/configuration#trace-log-targets" >}}) for the complete
matching, inheritance, and validation rules. The [Recording
Level](#recording-level) section below explains how the selected level affects
the recorded data.
+Java and Python can configure the same selection with `TraceLogTarget` through
`TRACE_LOG_TARGETS`:
+
+{{< tabs >}}
+{{< tab "python" >}}
+
+```python
+from flink_agents.api.core_options import AgentConfigOptions
+from flink_agents.api.trace_log import TraceLogLevel, TraceLogTarget
+
+agents_env.get_config().set(
+ AgentConfigOptions.TRACE_LOG_TARGETS,
+ [
+ TraceLogTarget.all(TraceLogLevel.STANDARD),
+ TraceLogTarget.tool("search", TraceLogLevel.VERBOSE),
+ ],
+)
+```
+
+{{< /tab >}}
+{{< tab "java" >}}
+
+```java
+import java.util.List;
+import org.apache.flink.agents.api.configuration.AgentConfigOptions;
+import org.apache.flink.agents.api.logger.TraceLogLevel;
+import org.apache.flink.agents.api.logger.TraceLogTarget;
+
+agentsEnv.getConfig().set(
+ AgentConfigOptions.TRACE_LOG_TARGETS,
+ List.of(
+ TraceLogTarget.all(TraceLogLevel.STANDARD),
+ TraceLogTarget.tool("search", TraceLogLevel.VERBOSE)));
+```
+
+{{< /tab >}}
+{{< /tabs >}}
+
+`TraceLogTarget` provides `all()` and whole-type factories `events()`,
`actions()`, `tools()`, `llms()`, `parsers()`, and `subagents()`. The
corresponding singular factories accept a name, such as `event("OrderCreated")`
or `tool("search")`. All factories accept an optional `TraceLogLevel`. Omitting
it preserves inheritance: `tool("search")` inherits the level of a matching
Tool type target or `ALL`, as described below. Targets are immutable.
+
+#### Choosing scopes and names
+
+Each target combines a `scope` with an optional `level`. In YAML, `EVENT_ONLY`
selects Events, and `ALL` selects all entity types the framework observes. Use
a category selector to narrow the selection to a named entity, such as `{tool:
search}`. The category keys are `event`, `action`, `tool`, `llm`, `parser`, and
`subagent`. A value of `"*"` selects the whole type: `{event: "*"}` is
equivalent to `EVENT_ONLY`.
+
+For a custom entity type reported through `ExecutionReporter`, use the generic
form `{entityType: custom, entityName: name}`; omit `entityName` to select the
whole type. `TraceLogTarget.entity("retriever", "catalog")` makes the same
selection as `scope: {entityType: retriever, entityName: catalog}`.
+
+For `entityName`, use the logical name from your Agent's code or
configuration. Trace Log records and target selectors use the same name:
+
+| `entityType` | What to use as `entityName` | Example |
+|--------------|----------------------------|---------|
+| `event` | The Event type used when emitting or subscribing to Events. |
`OrderCreated` |
+| `action` | The name used to declare the Action in the Agent. |
`create_order` |
+| `llm` | The registered ChatModel Resource name used by the call. |
`primary_model` |
+
+For example, `scope: {llm: primary_model}` selects calls to the Resource named
`primary_model`. Its configured model or deployment is recorded separately in
`entityMetadata.model`.
+
+Entity names match exactly. A name ending in `.*` matches its namespace and
descendants: `{event: "com.foo.*"}` matches `com.foo` and `com.foo.Event`, but
not `com.foobar.Event`. Other wildcard forms are unsupported.
+
+Direct model or tool calls inside an Action are not automatically recorded as
separate calls, even with `scope: ALL`. Selecting an entity does not
automatically select its parents or children. For example, selecting a Tool
does not select its parent Action; recorded relationships retain their IDs even
when related records are omitted.
+
+#### Matching scopes and inheriting levels
+
+For each entity, matching scopes are considered from most to least specific:
+
+| Order | Matching scope |
+|-------|----------------|
+| 1 | Exact entity name within its type |
+| 2 | Namespace prefixes within its type, from longest to shortest |
+| 3 | Entity type (`EVENT_ONLY` is the `event` type scope) |
+| 4 | `ALL` |
+
+The first explicitly specified `level` in this chain applies. An omitted
`level` inherits from the next matching scope and does not override it. If at
least one scope matches but no matching scope specifies `level`, the entity
uses `STANDARD`. If no scope matches, no record is written. Target order does
not affect the result.
+
+An explicit `OFF` stops inheritance and suppresses recording. A more specific
scope can enable recording by explicitly setting `STANDARD` or `VERBOSE`;
omitting `level` continues to inherit `OFF`. In YAML, write `level: "OFF"` with
quotes so Python does not interpret it as a boolean.
+
+For example, a Tool type target at `VERBOSE` supplies the level to matching
namespace and exact-name targets that omit it. A matching namespace target at
`OFF` suppresses an exact-name target that omits `level`, while an exact-name
target with explicit `STANDARD` or `VERBOSE` enables that entity. `EVENT_ONLY`
applies only to Events; its level does not affect a Tool target.
+
+An `ALL` target at `OFF` can provide a disabled baseline, with more specific
targets explicitly enabling recording. It is unnecessary when selecting only
one entity type. To record only Tools:
+
+```yaml
+agent:
+ trace-log:
+ targets:
+ - scope: {tool: "*"}
+ level: VERBOSE
+```
+
+To record everything except one Tool, use a local `OFF` target:
+
+```yaml
+agent:
+ trace-log:
+ targets:
+ - scope: ALL
+ level: STANDARD
+ - scope: {tool: search}
+ level: "OFF"
+```
+
+For accepted target shapes, duplicate-target validation, and all observability
options, see the [Configuration reference]({{< ref
"docs/operations/configuration#observability" >}}).
### Output
-The default output is **SLF4J**. Select **file** output with
`trace-log.output-type: FILE` or by setting a non-empty `trace-log.base-dir`; a
configured base directory takes precedence over `output-type`.
+The default output is **SLF4J**. Select **file** output with
`trace-log.output-type: FILE` or by setting a nonblank `trace-log.base-dir`; a
configured base directory takes precedence over `output-type`.
**SLF4J (default).** Records go to the logger category
`org.apache.flink.agents.TraceLog`. In Flink's default log4j2 setup, where
`log.file` is set, Flink Agents adds an appender that writes them as plain JSON
to `{log.file}.trace-log.log`; in the Flink Web UI, the file is listed in the
TaskManager's **Log List**, and it can be passed directly to the [Trace Tree
reader](#trace-tree-reader). An existing configuration for this logger is
respected. If no dedicated logger is configured and ` [...]
@@ -316,6 +419,10 @@ For an Event, `attributes` contains its payload and
`entityMetadata` its ID and
`producerExecutionId` identifies the Action execution that produced the Event,
`upstreamEventId` the preceding Event in its flow, and `upstreamActionName` the
associated Action. A runtime-generated Event may have an upstream Event without
an Action producer. Action records carry `triggerEventId`, linking the
execution to the Event that triggered it. When an Action is not selected, its
selected component calls still keep their `parentExecutionId`.
+When recovery reuses a completed Action, its saved output Events keep their
original
+Event IDs. Trace Log `entityMetadata` records their relationships to the
current
+trigger Event.
+
An LLM call uses its configured ChatModel Resource as `entityName`:
```json
