[rikkurea-boop]

Summary

Prevent LazyFileStorage.flush_step() from clearing the dirty state of a newer write that races with an older flush.

Problem

flush_step() snapshots the current dataframe under the storage lock, persists that snapshot outside the lock, then reacquires the lock and clears the step from _dirty_steps.

A concurrent write() may replace the buffered dataframe and mark the same step dirty after the old flush has taken its snapshot but before that flush completes.

The old flush would then unconditionally clear _dirty_steps, incorrectly marking the newer generation clean even though it has not been persisted.

Fix

Only clear the dirty marker when the dataframe currently stored for the step is still the same dataframe generation that was persisted by this flush.

Conceptually:

with self._lock:
    if self._buffers.get(step) is df:
        self._dirty_steps.discard(step)

[rikkurea-boop]

This fix has been consolidated into #550, together with the related batched storage changes, to simplify review.

The combined PR also addresses the overlapping-flush ordering regression and passed all six CI checks.

Closing this PR in favor of #550. Thanks!