Function transaction_reaper_completion_decision 

pub fn transaction_reaper_completion_decision(
    current: TransactionSnapshot,
    prepared: TransactionSnapshot,
    completion: TransactionSnapshot,
    exact_prepared_snapshot: bool,
) -> TransactionReaperCompletionDecision

Recheck a prepared snapshot after the abort- or commit-marker fan-out.

This is transaction_completion_decision with two additions: every snapshot must name a nonnegative producer identity, because the live entry and the prepared one come from the replayed transaction log; and exact_prepared_snapshot must hold for Proceed. The host computes it from equality over the complete persisted transaction entry, including its staged identity, partition set, timeout, and timestamps, so a registration or recovery that kept the identity and state still blocks the completion.

prepared.state and completion.state are the prepare and complete tags of one Kafka completion, which the host takes from a fixed Prepare* -> Complete* pairing, so they differ.