I think the issue here could be that the element of a stateful DoFn is a tuple[key, value]. One state per key/window, but since you're using global windows, there's one state per key.

Your BagStateSpec parameter stores only the value part of element. But as mentioned above, since you're storing only a single value, you might want to switch to ReadModifyWriteStateSpec.

So first: key, value = element

Consider also adding input type hints to your DoFn: @beam.typehints.with_input_types(KV[str, TimestampedValue])

More here: https://beam.apache.org/blog/stateful-processing/