Planning Flink Agents 1.0 - Brainstorming

[wenjin272]

Hi everyone,

With Flink Agents 0.4 targeting release this Friday, October 30, we'd like to start brainstorming what we should work on for Flink Agents 1.0. We plan to discuss this at the community sync on October 28, and welcome ideas here both before and after the meeting.

What would you like to see in 1.0?

Please share the use cases you want to build, pain points you've encountered, and features or improvements you think we should prioritize. Ideas around API stability and compatibility, production readiness, agent capabilities, integrations, developer experience, documentation, and examples are all welcome. Feel free to suggest other directions too!

For each idea, it would be helpful to include:

• The problem or use case and who would benefit.
• The proposed work and why it matters for 1.0.
• Related issues or discussions, if any.
• Interest in helping with design, implementation, testing, or documentation (optional).

A short comment is enough to get started; a detailed design is not required. You can also build on someone else's suggestion with your own use case or feedback.

This is an open brainstorm. We'll use the ideas and the sync discussion to shape the scope and priorities together; the 1.0 scope and release timeline are still to be discussed.

For context, the 0.4 planning discussion may provide useful starting points for follow-up work.

[sangkyoonnam]

For 1.0, I'd like to work on state compatibility tests across 1.x releases.

If the 1.0 compatibility policy covers state as well as APIs, a user should be able to take a savepoint on 1.0 and restore it on a later 1.x release without losing Flink-managed agent memory or pending inputs. I couldn't find a test that restores state written by an earlier release: RescalingTest and ActionStateSerializerRestoreTest restore what the same build wrote. Flink core keeps versioned operator snapshots in Git and restores them in tests such as WindowOperatorMigrationTest and CEPMigrationTest, using flink-migration-test-utils. I'd propose the same kind of test for the keyed and union list state in OperatorStateManager and DurableExecutionManager, following up on #1055's goal of defining the compatibility approach after 0.4. I can take the design and implementation.

Durable execution also needs an explicit upgrade contract. ActionStateSerde documents the action-state journal as checkpoint-scoped, with its memory-update payload guaranteed readable only with the same code and Flink version, so the policy would need to say whether in-flight actions survive an upgrade. apache/flink-web#888 already lists 0.3 action-state data as incompatible with 0.4.

Should state compatibility be part of the 1.0 policy?

[wenjin272]

Thanks for the experiments! My thoughts on the three questions:

1. The compatibility guarantee should cover upgrades within 1.x, excluding 0.4 → 1.0.
2. We should support snapshots with pending actions and events without requiring a full drain. Long-running actions could make draining impractical or cause checkpoint timeouts. We already persist pending work in state; the goal is to make its restoration and execution compatible across versions.
3. Agreed—we should expose a way to set a stable operator UID through the Agent API before 1.0.

[sangkyoonnam]

Thanks. I'll open an issue to let users set a stable operator UID through both the Java and Python agent APIs. Without an explicit UID, Flink derives the operator ID from the stream topology, and some topology changes can alter it. For an agent operator without one, KafkaActionStateCleanupTool plan currently requires --operator-uid-hash to read its recovery markers from a snapshot.

I'd also like to take #1332 for the cross-release test setup and its first at-rest scenario, then follow up here with a design and restore scenarios for pending actions and events. Could you assign #1332 to me?

[wenjin272]

@sangkyoonnam A clarification to my earlier reply: after discussing this further, we don’t think ActionState compatibility across versions should be guaranteed. ActionState is intended to avoid repeating recorded work during failure recovery with the same code and runtime versions. For planned upgrades, we can restore from a savepoint without reusing the old ActionState journal, accepting that calls within unfinished actions may execute again.

[sangkyoonnam]

I tried the upgrade path from your clarification with the Kafka store, and on 0.4 I couldn't find a way to restore a savepoint without reusing the old ActionState journal while a store stays configured.

With an operator test harness against a real Kafka broker (testcontainers, on main and on release-0.4.0-rc1), I took a checkpoint, completed one action, and in separate same-version runs restored the checkpoint four ways, resubmitting the input after each successful restore:

• Same topic: the restore replays the old journal, so the action isn't called again.
• New topic: the restore fails in KafkaActionStateStore.validateVersionedMarker with "recovery marker references topic …, but the configured topic is …".
• Same topic, deleted and recreated: the restore fails with "… the topic may have been recreated".
• No store: the restore works and the action runs again, but the upgraded job has no durable execution.

The checks are deliberate, since they keep a job from silently losing records, and DurableExecutionManager.handleRecovery passes the restored markers to the store whenever one is configured. So for these 0.4 Kafka markers, the routes left are reusing the journal or dropping the store. I haven't tried a 0.3 savepoint, whose markers are plain offset maps, or the Fluss store.

One option is an explicit upgrade mode that replaces the old journal's recovery markers and, where cleanup is configured, its cleanup metadata. It would have to keep replaying from the new journal on later failures, including any before the first new checkpoint. Would something like that fit the 1.0 plan, or would you rather document a different procedure?

[wenjin272]

Thanks for testing this! An explicit upgrade mode sounds like the right direction: skip the old journal while keeping ActionStateStore enabled for subsequent failure recovery.

[purushah]

I’d consider context compaction critical for long-running agents in 1.0. Tool results and conversation history can keep growing until they exceed the model’s context limit, as we saw in our tests. We also saw incorrect answers before reaching that limit.

The goal should be to keep context manageable while preserving the information agents need to complete their tasks. #1202 covers the proposal and findings.

[wenjin272]

Sharing a proposed set of priorities for Flink Agents 1.0.0 from @xintongsong, @pltbkd, @joeyutong and @wenjin272, following our initial brainstorming discussions.

Must-have for 1.0:

• More control over model and tool execution. Let users insert custom logic around model/tool calls, capture intermediate results, and intervene in the built-in ReAct loop. Hooks, events, and explicit tool-call handling are options to evaluate against these requirements.
• Agent loop budget controls. Support time, round/step, and token budgets, with options to stop immediately or allow a final round for the model to wrap up when a budget is reached.
• Timer support for delayed tasks. Allow agents to schedule future events using Flink timers, enabling delayed checks and timeout handling without keeping an action running while waiting.
• Distinguishing responses from multiple requests. Help downstream actions distinguish responses from multiple requests and handle them appropriately by propagating user-defined context from each request to its response.
• Decoupling checkpointing from agent execution. Agent runs may involve multiple rounds of model inference, tool calls, and subagent execution with unpredictable latency. Allow checkpoints to complete without waiting for these runs to finish, and define how in-progress execution is saved and recovered.
• API, configuration, and built-in integration compatibility. Define stability labels and compatibility rules for APIs, configuration, YAML, and built-in integrations, including deprecation and migration. Add annotations/decorators and automated compatibility checks in CI.
• State compatibility. Framework upgrades should preserve operator state when application logic and state usage remain unchanged, allowing necessary API adaptations. Define the supported version window and validate it with cross-release restoration tests. Compatibility after changes to application logic would require separate evaluation. Cross-version reuse of the ActionState journal would not be guaranteed; calls within unfinished actions may execute again after an upgrade.
• Framework support for multimodal data. Complete the abstractions needed for multimodal inputs, outputs, and tool results. Clarify which capabilities fit existing model interfaces and which need another abstraction; individual model integrations can follow separately.
• Rename Flink Agents. We have repeatedly seen people interpret the name as an agent for operating Flink, rather than an agent framework built on Flink. Consider a clearer name for 1.0 that better communicates the project’s purpose.

Nice to have:

• Stage-level quality gates — high priority. Support checks on output quality, evidence, or executability, followed by bounded retries or a return to the relevant stage while retaining successful work.
• Unified serial and parallel execution paths — high priority. Consolidate duplicated runtime logic and align lifecycle and event-listener behavior. Evaluate concurrency and backpressure separately, favoring clear limits on in-flight inputs and asynchronous tasks where additional control is needed.
• Stronger structured-output validation. Extend schema and field validation, including nullability constraints and handling of validation failures. Keep this separate from stage-level quality gates.
• Application versus framework failures. Clarify which tool/subagent failures can become responses for the caller to handle, and which must propagate to the runtime and trigger recovery.
• Configurable persistence of failed calls. Avoid repeatedly replaying a recorded transient failure that causes the job to fail again. The proposed default is to leave failed durable calls unrecorded, with an explicit per-call option when recording the failure is required. The precise failure boundary still needs design.
• Human-in-the-loop abstractions. Simplify approval, rejection, and resumption workflows. Timer support is the more urgent framework building block; application-specific approval flows can be assembled with custom actions and external systems.
• Prompt hot updates. Explore refreshing prompts from MCP, remote files, or prompt-management services without replacing the entire agent plan.
• Python and YAML Model Routing support. Fill the remaining API gaps relative to Java.
• Support for Flink batch mode. Enable Flink Agents to run in Flink batch mode. Investigate asynchronous execution and key-context lifecycle constraints, then define the supported scope.
• ActionStateStore improvements. Improve testing, identity handling, cleanup, write-failure handling, and storage efficiency. A broader redesign needs a clearer use case and scope.
• Cross-language external subagents. Address the remaining Java/Python interoperability gaps and clarify their relationship to internal subagent support.
• Operational visualization. Explore a graph of observed execution relationships, combined with metrics such as event counts or rates. This should summarize runtime behavior and complement individual execution traces. Metric coverage and the UI integration approach need further design.
• Additional multimodal model integrations. Expand model coverage based on concrete use cases, building on the framework-level multimodal abstractions.
• CDC integration. Explore invoking agents from CDC pipelines, with scope and timing informed by concrete use cases and the Flink CDC release plan.

Longer-term directions and upstream work that should not block 1.0:

• Subagents on RPC operators. Explore delegating subagent execution to separate worker resources, subject to concrete use cases and upstream Flink support.
• Full AgentPlan hot updates. Support updating agent plans, including plan, JAR, and dependency persistence and recovery. Continue the upstream groundwork while evaluating prompt-only updates separately.
• SQL integration. Advance the Flink proposal, distinguishing calls to existing agents from declaring and reusing YAML agents through SQL/Catalog.
• New input during ongoing reasoning — high priority for longer-term exploration. Explore how a new event for the same key can supplement or interrupt ongoing reasoning while preserving well-defined state and consistency semantics.

Development and release practices:

• Documentation alongside features. Require relevant documentation updates alongside changes to user-visible behavior, APIs, or configuration, and review them together before merging.
• Documentation checks. Record this expectation in contributor and reviewer guidance. Explore automated checks and periodic audits to help catch omissions.
• Routine release preparation. Keep release testing in routine CI, maintain a releasable main branch, and continue automating release-note preparation.
• Smaller, more frequent releases. Keep release-blocking requirements focused and reduce accumulated release work. A cadence of roughly two weeks to one month is a direction to work toward, rather than a committed schedule. Larger documentation reorganizations can be handled as separate contributions.

[wenjin272]

Adding two more ideas from yesterday’s community sync brainstorming for 1.0.0:

• Provide APIs to delete entries from sensory memory and short-term memory.
• Consider renaming short-term memory. In Flink Agents, this memory is also persisted, so “short-term” may give users the impression that it is temporary or not durable. A clearer name could help avoid this confusion.

These are still open for discussion. Suggestions on the APIs and naming are welcome!