GitHub user Ashfaqbs added a comment to the discussion: Design proposal for 
#1066: dry-run / replay via savepoints

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.

GitHub link: 
https://github.com/apache/flink-agents/discussions/1069#discussioncomment-18212480

----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]

Reply via email to