[joeyutong]

Purpose of change

ActionExecutionOperator had two prune-state boundary issues around checkpoints.

• prune bookkeeping used messageSequenceNumber, which tracks the latest started sequence, not the latest completed sequence. That means an in-flight sequence could be treated as prune-safe.

• the operator called pruneState as soon as a run completed, before that completion was covered by a completed checkpoint.

Both issues are operator-side prune-boundary problems. In the current Kafka backend, pruneState only evicts the in-memory cache, so the correctness impact is mostly masked by replay/rebuild from the durable log. Still, the operator should not rely on that backend detail: the state-store prune boundary should only include completed sequences that are part of a completed checkpoint.

Fix

This change makes pruning checkpoint-safe again.

• track a separate lastCompletedSequenceNumber keyed state
• snapshot prune metadata from lastCompletedSequenceNumber instead of messageSequenceNumber
• remove the eager prune path on run completion
• prune action state only after notifyCheckpointComplete

Tests

The updated tests cover:

• no pruning before checkpoint completion
• no pruning of sequences that are still in flight
• replay from an earlier checkpoint keeps durable state available
• action state cleanup happens after checkpoint completion

API

No

Documentation

• doc-needed
• doc-not-needed
• doc-included

[xintongsong]

LGTM