Design proposal for #1066: dry-run / replay via savepoints

[Ashfaqbs]

Following up on #1066 since @wenjin272 asked for a concrete design before this goes anywhere. Putting a rough shape here to argue about, not committing to build it myself.

The gap

Right now the only way to know how a change to an agent (new prompt, different model, tweaked parameters, changed Action logic) will actually behave is to ship it and watch. There's no way to run it against state the agent has already built up and see what happens first.

This is a decent fit for Flink Agents specifically because agent state is already durable and keyed, so a savepoint is a real, consistent snapshot to replay against - not something that needs to be built from scratch for this.

Rough shape

The core idea: a replay job is a regular Flink Agents job started from a savepoint, but reading from a recorded input stream instead of live traffic, with external side effects intercepted instead of fired.

That breaks down into three parts.

Capturing the input. An opt-in sink attached to the existing input stream that records the keyed events in order (event time, key, payload) so there's something deterministic to play back later. Doesn't touch runtime state - it just sits next to the existing pipeline.

Running the replay. Flink already lets you start a job from a savepoint (bin/flink run -s :savepointPath). Point that at a ReplaySource reading the recorded stream instead of the real source, keeping per-key ordering and event-time so watermarks and keyed state behave the same as the original run.

Containing side effects. This is the part I'd want the most pushback on. Actions that call out to a ChatModel or a tool need to behave differently in replay - either the call gets redirected somewhere inspectable instead of firing, or (if the whole point is testing a new prompt against real state) the call is allowed but anything not explicitly marked safe to repeat gets blocked. My instinct is a ReplayContext reachable from RunnerContext, defaulting to "don't execute, just record what would have happened" unless an Action opts in as idempotent/read-only.

What it shouldn't need

No changes to checkpoint or recovery semantics - from Flink's point of view a replay job is a normal job, just pointed at a different source. No new keyed state either; the capture sink and the savepoint restore both reuse what already exists.

Where I'm not sure

• Should side-effect safety be marked on the Action itself (a decorator) or kept in a separate registry? Decorator is more discoverable, but it ties a runtime concern to the Action definition.
• Does the capture sink belong in core, or is it thin enough to just document as a pattern (a sink plus a source) without adding new API surface?
• For a first cut, is comparing old-vs-new output on the same replayed input in scope, or is "run once against real state and look at it" enough to start with?

No illusions about timing given the 0.4 freeze on Sept 15 - just want the shape settled for whenever it does get picked up.

[weiqingy]

Thanks for putting this together. Savepoint-based replay seems like a promising direction, and Flink already provides a consistent snapshot of the agent’s keyed state. After looking through the current runtime, I wonder if a few recovery and isolation details need to be defined more precisely.

How would the agent state map to the replay topology after replacing the live source? Flink restores state by operator ID, but the current action-execute-operator has no explicit UID or max parallelism. Would replay support first require stable operator identities and an explicit rule for dropping the old source state?

How would a savepoint identify the exact starting point in the captured input? Flink’s exactly-once state guarantee relies on restoring both operator state and the matching source position. Since ReplaySource replaces the original source, would the capture path need a checkpoint-linked manifest containing the replay offsets, typed key information, serializers, and key ordering? If reproducing downstream event-time behavior is also a goal, would watermarks and source partition information need to be included as well?

I also wonder whether an ordinary savepoint contains too much execution state for this use case. Flink Agents checkpoints pending ActionTasks, pending input events, processing keys, sequence numbers, and durable recovery markers in addition to memory. A pending task also contains the Action definition from the original job. Could a replay of changed Action logic therefore resume old logic first? Would a framework-specific quiesce-and-snapshot flow, or a derived savepoint containing only replay-safe state, provide a clearer boundary?

How should the external action-state store be handled? Restoring the savepoint rebuilds action state from the configured Kafka topic or Fluss table, while new replay actions write back to that same store. Because the store keys include the business key and sequence number but no replay namespace, could running production and replay together create divergent recovery records? Would replay need a separate store with restored recovery markers removed or rewritten?

Long-term memory has a similar question. The restored job identifier points to the production Mem0 namespace. A new identifier would isolate replay writes, but it would no longer expose the production long-term memory being evaluated. Would a read-only snapshot or fork of that namespace be needed?

Finally, can an Action-level idempotent/read-only marker enforce the intended safety? Custom Actions can call HTTP services, databases, or filesystems directly, outside framework-managed resources. Would it be clearer to enforce replay policy at ChatModel, Tool, vector-store, and long-term-memory boundaries, while treating direct external calls as unsupported unless the replay job is sandboxed?

Perhaps a focused first version could combine stable operator IDs, a savepoint-linked input manifest, replay-safe state projection, isolated external state, and replay-aware resource proxies. Would that give the feature a safer and more testable starting point?

[Ashfaqbs]

Thanks for the detailed read-through — every one of these is a real gap, not a hypothetical. Went and checked each against the current runtime rather than answering from the proposal alone, so here's where things actually stand and how I'd narrow the v1 in response.

Operator identity. Confirmed: CompileUtils.connectToAgent() never calls .uid(...) or .setMaxParallelism(...) on the action-execute-operator — both are left to Flink's topology-derived defaults. That's fine for restart-in-place today because the topology never changes, but it's exactly the kind of thing that breaks silently the moment a replay job swaps the source. Agreed this needs an explicit, stable UID (and pinned max parallelism) as a prerequisite, not something replay can route around.

Savepoint-to-input mapping. There's no offset/position manifest anywhere in the codebase today — a savepoint is just Flink-managed operator/keyed state, with nothing tying it to a specific point in the source. So yes, the capture path would need to write a checkpoint-linked manifest (offsets, key schema, serializers, ordering, and watermark/partition info if we want event-time behavior to reproduce) rather than treating the savepoint alone as sufficient.

Savepoint scope / stale Action logic. This is the sharpest one. ActionTask objects held in pending state carry the actual Action from the AgentPlan compiled at snapshot time. So a savepoint taken mid-flight, restored under a newer build with changed Action logic, would finish those in-flight tasks with the old logic before anything new ever runs — silently, since nothing surfaces that mismatch today. I don't think an ordinary savepoint is the right unit for this at all; it sounds like replay needs either a quiesce-and-snapshot flow (drain in-flight ActionTasks to a clean boundary before capturing) or a derived/filtered savepoint that only carries state guaranteed replay-safe (committed action results, memory, no in-flight tasks). Quiescing seems more tractable than trying to define "replay-safe state" as a general concept.

Action-state store namespacing. Also confirmed as described: ActionStateUtil.generateKey() builds keyGroup:seqNum:eventUUID:actionUUID:businessKey, with no replay/job dimension. Since the store's topic/table comes from static agent config, a replay job using the same config as production would write into the same keyspace — sharing a topic with the source it's replaying against invites exactly the divergent-record problem you're describing. I don't see a way to fix this by change to the key format alone (it's used for prefix/ownership matching elsewhere) — replay needs its own store, seeded from the restored prod state, with production write access severed.

Long-term memory isolation. Confirmed, and worse than "how should this be handled" — it's currently the opposite of isolated. ActionExecutionOperator deliberately restores jobIdentifier from operator state specifically so it survives a savepoint/job-ID change (there's a comment citing exactly that scenario). So today, restoring a savepoint under a new job automatically keeps writing to the same Mem0 namespace as the original job — replay would need to deliberately override this, and as you say, a fresh identifier means losing visibility into the very memory state being evaluated. A read-only fork/snapshot of the Mem0 namespace at restore time seems like the only option that doesn't trade one problem for the other.

Enforcement boundary. Agreed on scoping the guarantee to what the framework actually mediates. ActionExecutionOperator.open() resolves every ChatModel/Tool/vector-store/LTM instance through a single ResourceCache — that's the one real interception point today. A custom Action making a raw HTTP/DB/filesystem call bypasses it entirely and is invisible to the runtime, so there's no honest way to "enforce" replay-safety there without sandboxing the whole job (network policy / container isolation). I'd rather the framework be explicit that unmediated external calls are out of scope for v1 than imply a guarantee it can't back up.

Given all of this, the v1 shape I'd argue for is narrower than what I first sketched: stable operator UID + pinned max parallelism, a checkpoint-linked input manifest, quiesce-before-snapshot instead of an ordinary savepoint, an isolated action-state store for replay, a forked/read-only Mem0 namespace, and replay policy enforced only at the ResourceCache boundary with direct external calls documented as unsupported outside a sandboxed job. That's a lot of prerequisite plumbing before "replay" itself is even the interesting part — makes sense why this needed a design pass before code.