Function upsert_source_offset_hashed 

pub fn upsert_source_offset_hashed(
    txn: &mut ApplyTxn,
    source: &str,
    offset: &str,
    chunk_seq: i64,
    run_id: Option<&str>,
    batch_hash: Option<&str>,
) -> Result<(), ReplicaError>

upsert_source_offset carrying the §8.3 batch identity: the emitter-computed batch_hash of the run’s exact bytes, stored beside the cursor so a same-offset resubmission with a different body is a loud error, never a silent dedup (T6). The hash-less paths (follower stream, plain change sources) write NULL — once the cursor advances, the previous offset’s identity is dead anyway.