diff --git a/Cargo.lock b/Cargo.lock
index edac76b..4f540dd 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -19,6 +19,18 @@ dependencies = [
  "cpufeatures",
 ]
 
+[[package]]
+name = "ahash"
+version = "0.8.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75"
+dependencies = [
+ "cfg-if",
+ "once_cell",
+ "version_check",
+ "zerocopy",
+]
+
 [[package]]
 name = "anyhow"
 version = "1.0.104"
@@ -76,6 +88,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cb3028782f6bf14a6df987244333d34e6b272b5a40a53e4879ec2dfd82275a3a"
 dependencies = [
  "bitcoin",
+ "hashbrown",
+ "serde",
 ]
 
 [[package]]
@@ -90,6 +104,7 @@ dependencies = [
  "futures",
  "futures-timer",
  "miniscript",
+ "serde",
  "serde_json",
  "tokio",
  "tokio-util",
@@ -573,6 +588,16 @@ dependencies = [
  "r-efi",
 ]
 
+[[package]]
+name = "hashbrown"
+version = "0.14.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
+dependencies = [
+ "ahash",
+ "serde",
+]
+
 [[package]]
 name = "hex-conservative"
 version = "0.2.2"
diff --git a/bdk_electrum_streaming/Cargo.toml b/bdk_electrum_streaming/Cargo.toml
index 646d1ab..15e23fa 100644
--- a/bdk_electrum_streaming/Cargo.toml
+++ b/bdk_electrum_streaming/Cargo.toml
@@ -13,9 +13,10 @@ readme = "README.md"
 futures = "0.3"
 futures-timer = "3"
 anyhow = "1"
-bdk_core = "0.6"
+bdk_core = { version = "0.6", features = ["serde"] }
 miniscript = { version = "12.0.0" }
 electrum_streaming_client = { version = "0.4" }
+serde = { version = "1", features = ["derive", "rc"] }
 serde_json = "1"
 tracing = "0.1"
 
diff --git a/bdk_electrum_streaming/src/cache.rs b/bdk_electrum_streaming/src/cache.rs
new file mode 100644
index 0000000..7de9857
--- /dev/null
+++ b/bdk_electrum_streaming/src/cache.rs
@@ -0,0 +1,346 @@
+use std::{
+    collections::{BTreeMap, BTreeSet, HashMap, HashSet},
+    sync::Arc,
+};
+
+use bdk_core::{
+    bitcoin::{self, BlockHash, Transaction, Txid},
+    ConfirmationBlockTime,
+};
+use electrum_streaming_client::{response, ElectrumScriptHash, ElectrumScriptStatus};
+
+/// Everything learned from the server, kept so a reconnect need not ask again.
+#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
+pub struct Cache {
+    /// The server's per-script histories.
+    pub spk_histories: SpkHistories,
+    /// Every txid ever seen for each script hash.
+    ///
+    /// Stays here rather than in [`SpkHistories`] because it is the one spk-keyed map that must
+    /// survive [`SpkHistories::remove`]. Two things read it after the server has stopped
+    /// reporting a history: evictions are the difference between this set and the history now
+    /// in hand, so replacing it with the latest history would make that difference empty and
+    /// no transaction would ever be reported as evicted; and it is the record that a script
+    /// was *once* active, which keeps its derivation index revealed and the lookahead
+    /// extended past it.
+    pub spk_txids: HashMap<ElectrumScriptHash, BTreeSet<Txid>>,
+    pub txs: HashMap<Txid, Arc<Transaction>>,
+    /// Written as a sequence: a `(Txid, BlockHash)` key is not a string, so a map would be
+    /// unserializable in JSON and every other format that requires string keys.
+    #[serde(with = "persist::anchors_as_seq")]
+    pub anchors: HashMap<(Txid, BlockHash), ConfirmationBlockTime>,
+    pub headers: HashMap<BlockHash, bitcoin::block::Header>,
+}
+
+/// The last history the server reported for each script hash.
+///
+/// The only part of [`Cache`] a caller cannot rebuild from wallet data, since the server reports
+/// a script's history as it stands now and will never again mention a transaction it has dropped.
+///
+/// Fields are private: the height index is derived from the histories, and letting it be set
+/// independently would reintroduce the desync the type exists to prevent.
+#[derive(Debug, Clone, Default)]
+pub struct SpkHistories {
+    /// The last history reported for each script hash, with the status it stands for.
+    ///
+    /// The status is stored with the history rather than beside it so the two cannot desync: a
+    /// replay needs the last status, and [`Self::get`] needs to know which status the
+    /// history it hands back is an answer to.
+    spk_hash_to_history: HashMap<ElectrumScriptHash, (ElectrumScriptStatus, Vec<response::Tx>)>,
+    /// Script hashes whose history reported a transaction at each height.
+    ///
+    /// This is what makes a reorg actionable: when the local chain drops a block, the scripts
+    /// recorded at that height are the ones whose anchors need refetching. Derived from
+    /// `spk_hash_to_history`, so it is rebuilt on deserialization rather than stored.
+    height_to_spk_hashes: BTreeMap<u32, BTreeSet<ElectrumScriptHash>>,
+}
+
+impl SpkHistories {
+    /// How far below the tip the height index is retained by [`Self::prune`].
+    ///
+    /// Comfortably above [`ChainJob`]'s 21-block suffix, which bounds how deep an eviction — the
+    /// only thing that reads the index — can ever reach.
+    ///
+    /// [`ChainJob`]: crate::chain_job::ChainJob
+    pub const HEIGHT_INDEX_HORIZON: u32 = 100;
+
+    /// Drop the history for `spk_hash`, for when the server stops reporting one.
+    ///
+    /// Does not touch [`Cache::spk_txids`], which has to outlive this to report the evictions.
+    pub fn remove(&mut self, spk_hash: ElectrumScriptHash) {
+        self.spk_hash_to_history.remove(&spk_hash);
+        for spk_hashes in self.height_to_spk_hashes.values_mut() {
+            spk_hashes.remove(&spk_hash);
+        }
+    }
+
+    /// Record `history` as the answer to `spk_status`, replacing whatever `spk_hash` had before.
+    pub fn insert(
+        &mut self,
+        spk_hash: ElectrumScriptHash,
+        spk_status: ElectrumScriptStatus,
+        history: Vec<response::Tx>,
+    ) {
+        for tx in &history {
+            if let Some(height) = tx.confirmation_height() {
+                self.height_to_spk_hashes
+                    .entry(height.to_consensus_u32())
+                    .or_default()
+                    .insert(spk_hash);
+            }
+        }
+        self.spk_hash_to_history
+            .insert(spk_hash, (spk_status, history));
+    }
+
+    /// The history for `spk_hash`, but only if it is the one `spk_status` stands for.
+    ///
+    /// A job fetches for the status its notification carried, so handing it a history that
+    /// answers an older status would have it finish on stale data. `None` sends it to the
+    /// server instead, which is why the status is effectively part of the key.
+    pub fn get(
+        &self,
+        spk_hash: ElectrumScriptHash,
+        spk_status: ElectrumScriptStatus,
+    ) -> Option<&[response::Tx]> {
+        match self.spk_hash_to_history.get(&spk_hash)? {
+            (status, history) if *status == spk_status => Some(history),
+            _ => None,
+        }
+    }
+
+    /// Every script hash whose history reported a transaction at one of `heights`, deduplicated.
+    ///
+    /// Given the heights a reorg evicted, these are the scripts whose anchors need refetching.
+    pub fn spk_hashes_at_heights<'a>(
+        &'a self,
+        heights: impl IntoIterator<Item = u32> + 'a,
+    ) -> impl Iterator<Item = ElectrumScriptHash> + 'a {
+        heights
+            .into_iter()
+            .filter_map(|height| self.height_to_spk_hashes.get(&height))
+            .flatten()
+            .copied()
+            .filter({
+                let mut dedup = HashSet::new();
+                move |&spk_hash| dedup.insert(spk_hash)
+            })
+    }
+
+    /// The last status the server reported for `spk_hash`, if it still has a history.
+    pub fn status(&self, spk_hash: ElectrumScriptHash) -> Option<ElectrumScriptStatus> {
+        self.spk_hash_to_history
+            .get(&spk_hash)
+            .map(|&(status, _)| status)
+    }
+
+    /// Drop height index entries too far below `tip_height` for any reorg to reach.
+    ///
+    /// The histories themselves are untouched; only the index is bounded.
+    pub fn prune(&mut self, tip_height: u32) {
+        if let Some(height) = tip_height.checked_sub(Self::HEIGHT_INDEX_HORIZON) {
+            self.height_to_spk_hashes = self.height_to_spk_hashes.split_off(&height);
+        }
+    }
+}
+
+/// Types and impls that exist only so [`Cache`] can be stored and loaded.
+///
+/// Kept apart from the cache itself because [`HistoryTx`] mirrors [`response::Tx`] and the two
+/// are easy to mistake for each other at a glance.
+mod persist {
+    use super::*;
+
+    /// A history entry in the shape we can write back out.
+    ///
+    /// [`response::Tx`] derives `Deserialize` only, so histories round-trip through this instead.
+    #[derive(serde::Serialize, serde::Deserialize)]
+    enum HistoryTx {
+        Mempool {
+            txid: Txid,
+            fee_sats: u64,
+            confirmed_inputs: bool,
+        },
+        Confirmed {
+            txid: Txid,
+            height: u32,
+        },
+    }
+
+    impl From<&response::Tx> for HistoryTx {
+        fn from(tx: &response::Tx) -> Self {
+            match tx {
+                response::Tx::Mempool(tx) => Self::Mempool {
+                    txid: tx.txid,
+                    fee_sats: tx.fee.to_sat(),
+                    confirmed_inputs: tx.confirmed_inputs,
+                },
+                response::Tx::Confirmed(tx) => Self::Confirmed {
+                    txid: tx.txid,
+                    height: tx.height.to_consensus_u32(),
+                },
+            }
+        }
+    }
+
+    impl TryFrom<HistoryTx> for response::Tx {
+        type Error = bitcoin::absolute::ConversionError;
+
+        fn try_from(tx: HistoryTx) -> Result<Self, Self::Error> {
+            Ok(match tx {
+                HistoryTx::Mempool {
+                    txid,
+                    fee_sats,
+                    confirmed_inputs,
+                } => Self::Mempool(response::MempoolTx {
+                    txid,
+                    fee: bitcoin::Amount::from_sat(fee_sats),
+                    confirmed_inputs,
+                }),
+                HistoryTx::Confirmed { txid, height } => Self::Confirmed(response::ConfirmedTx {
+                    txid,
+                    height: bitcoin::absolute::Height::from_consensus(height)?,
+                }),
+            })
+        }
+    }
+
+    /// Only the histories are written; the height index is rebuilt from them on the way back in.
+    impl serde::Serialize for SpkHistories {
+        fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+            serializer.collect_map(self.spk_hash_to_history.iter().map(
+                |(&spk_hash, (status, history))| {
+                    (
+                        spk_hash,
+                        (
+                            status,
+                            history.iter().map(HistoryTx::from).collect::<Vec<_>>(),
+                        ),
+                    )
+                },
+            ))
+        }
+    }
+
+    impl<'de> serde::Deserialize<'de> for SpkHistories {
+        fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+            use serde::de::Error;
+            let stored =
+                HashMap::<ElectrumScriptHash, (ElectrumScriptStatus, Vec<HistoryTx>)>::deserialize(
+                    deserializer,
+                )?;
+            let mut spk_histories = Self::default();
+            for (spk_hash, (status, history)) in stored {
+                let history = history
+                    .into_iter()
+                    .map(response::Tx::try_from)
+                    .collect::<Result<Vec<_>, _>>()
+                    .map_err(D::Error::custom)?;
+                spk_histories.insert(spk_hash, status, history);
+            }
+            Ok(spk_histories)
+        }
+    }
+
+    pub(super) mod anchors_as_seq {
+        use super::*;
+        use serde::{Deserialize, Deserializer, Serializer};
+
+        type Anchors = HashMap<(Txid, BlockHash), ConfirmationBlockTime>;
+
+        pub fn serialize<S: Serializer>(
+            anchors: &Anchors,
+            serializer: S,
+        ) -> Result<S::Ok, S::Error> {
+            serializer.collect_seq(
+                anchors
+                    .iter()
+                    .map(|(&(txid, block_hash), anchor)| (txid, block_hash, anchor)),
+            )
+        }
+
+        pub fn deserialize<'de, D: Deserializer<'de>>(
+            deserializer: D,
+        ) -> Result<Anchors, D::Error> {
+            Ok(
+                Vec::<(Txid, BlockHash, ConfirmationBlockTime)>::deserialize(deserializer)?
+                    .into_iter()
+                    .map(|(txid, block_hash, anchor)| ((txid, block_hash), anchor))
+                    .collect(),
+            )
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use bitcoin::hashes::Hash;
+
+    fn txid(byte: u8) -> Txid {
+        Txid::from_byte_array([byte; 32])
+    }
+
+    fn spk_hash(byte: u8) -> ElectrumScriptHash {
+        ElectrumScriptHash::from_byte_array([byte; 32])
+    }
+
+    /// The height index is not stored, so it has to come back from the histories themselves.
+    #[test]
+    fn spk_histories_round_trip_rebuilds_the_height_index() {
+        let history = vec![
+            response::Tx::Confirmed(response::ConfirmedTx {
+                txid: txid(1),
+                height: bitcoin::absolute::Height::from_consensus(700_000).unwrap(),
+            }),
+            response::Tx::Mempool(response::MempoolTx {
+                txid: txid(2),
+                fee: bitcoin::Amount::from_sat(1234),
+                confirmed_inputs: false,
+            }),
+        ];
+        let status = ElectrumScriptStatus::from_history(&history).expect("history is not empty");
+
+        let mut before = SpkHistories::default();
+        before.insert(spk_hash(9), status, history);
+
+        let json = serde_json::to_string(&before).expect("must serialize");
+        assert!(
+            !json.contains("height_to_spk_hashes"),
+            "the derived index must not be stored"
+        );
+        let after: SpkHistories = serde_json::from_str(&json).expect("must deserialize");
+
+        assert_eq!(
+            after.get(spk_hash(9), status).map(|h| h.len()),
+            Some(2),
+            "the history must survive, and still answer to its status"
+        );
+        assert_eq!(
+            after.spk_hashes_at_heights([700_000]).collect::<Vec<_>>(),
+            vec![spk_hash(9)],
+            "the confirmed entry must be indexed by height again"
+        );
+        assert_eq!(
+            after.spk_hashes_at_heights([700_001]).count(),
+            0,
+            "only heights the history actually reported"
+        );
+        assert_eq!(after.status(spk_hash(9)), Some(status));
+    }
+
+    /// `anchors` is keyed by a tuple, which JSON cannot use as a map key.
+    #[test]
+    fn cache_round_trips_through_json() {
+        let anchor = (txid(1), bitcoin::BlockHash::from_byte_array([2; 32]));
+        let mut before = Cache::default();
+        before
+            .anchors
+            .insert(anchor, ConfirmationBlockTime::default());
+
+        let json = serde_json::to_string(&before).expect("must serialize");
+        let after: Cache = serde_json::from_str(&json).expect("must deserialize");
+
+        assert_eq!(after.anchors.get(&anchor), before.anchors.get(&anchor));
+    }
+}
diff --git a/bdk_electrum_streaming/src/chain_job.rs b/bdk_electrum_streaming/src/chain_job.rs
index e11d156..f03c27c 100644
--- a/bdk_electrum_streaming/src/chain_job.rs
+++ b/bdk_electrum_streaming/src/chain_job.rs
@@ -16,10 +16,35 @@ use std::collections::{BTreeMap, BTreeSet};
 /// [`process_blocks()`]: ChainJob::process_blocks
 #[derive(Debug, Clone)]
 pub struct ChainJob {
+    /// The block this job was created to reach.
+    ///
+    /// A `blockchain.block.headers` batch is answered from whichever chain the server holds when
+    /// it *replies*, which need not be the one it announced. Keeping the target lets
+    /// [`try_finish()`] tell the two apart instead of writing an unannounced chain into the
+    /// checkpoint chain.
+    ///
+    /// [`try_finish()`]: ChainJob::try_finish
+    target: BlockId,
     missing_headers: BTreeSet<u32>,
     cp_update: BTreeMap<u32, BlockHash>,
 }
 
+/// The outcome of [`ChainJob::try_finish`].
+#[derive(Debug)]
+pub enum ChainJobOutcome {
+    /// Every header arrived, and they agree with the block the job was created for.
+    Finished(CheckPoint),
+    /// Still waiting on headers.
+    Pending(ChainJob),
+    /// The headers arrived but disagree with the block the job was created for, so they describe
+    /// a chain other than the one announced.
+    ///
+    /// The job is abandoned rather than retried: the server only answers from a chain it has
+    /// moved to, and moving is what makes it announce a new tip, so the notification that
+    /// rebuilds the job is already on its way.
+    Superseded,
+}
+
 impl ChainJob {
     const CHAIN_SUFFIX_LENGTH: u32 = 21;
 
@@ -32,6 +57,10 @@ impl ChainJob {
         header: Header,
         height: u32,
     ) -> Option<Self> {
+        let target = BlockId {
+            height,
+            hash: header.block_hash(),
+        };
         let cp = local_tip
             .iter()
             .find(|cp| cp.height() <= height)
@@ -46,6 +75,7 @@ impl ChainJob {
                 if let Some(prev_height) = height.checked_sub(1) {
                     if prev_height == prev_cp.height() && header.prev_blockhash == prev_cp.hash() {
                         return Some(Self {
+                            target,
                             missing_headers: BTreeSet::new(),
                             cp_update: core::iter::once((height, header.block_hash())).collect(),
                         });
@@ -68,6 +98,7 @@ impl ChainJob {
                 count,
             });
             Some(Self {
+                target,
                 missing_headers: (start_height..=remote_height).collect(),
                 cp_update: BTreeMap::new(),
             })
@@ -82,6 +113,7 @@ impl ChainJob {
                 count: (remote_height + 1 - remote_start_height) as usize,
             });
             Some(Self {
+                target,
                 missing_headers: (local_start_height..=local_height)
                     .chain(remote_start_height..=remote_height)
                     .collect(),
@@ -105,13 +137,30 @@ impl ChainJob {
         self
     }
 
-    pub fn try_finish(self, local_tip: &mut CheckPoint) -> Result<CheckPoint, Self> {
+    pub fn try_finish(self, local_tip: &mut CheckPoint) -> ChainJobOutcome {
         if !self.missing_headers.is_empty() {
             tracing::trace!(
                 missing = self.missing_headers.len(),
                 "Chain job not finished"
             );
-            return Err(self);
+            return ChainJobOutcome::Pending(self);
+        }
+
+        // The batch is only evidence about the chain it was answered from. If it does not put
+        // the announced block at the announced height, that is a chain we were never told about
+        // and must not adopt — the same reasoning the merkle and header handlers apply to a
+        // proof built against a block other than ours.
+        if self.cp_update.get(&self.target.height) != Some(&self.target.hash) {
+            tracing::info!(
+                height = self.target.height,
+                announced = self.target.hash.to_string(),
+                received = self
+                    .cp_update
+                    .get(&self.target.height)
+                    .map(|h| h.to_string()),
+                "Headers describe a chain other than the one announced. Abandoning chain job.",
+            );
+            return ChainJobOutcome::Superseded;
         }
 
         let mut cp = local_tip.clone();
@@ -124,6 +173,6 @@ impl ChainJob {
             tip_hash = cp.hash().to_string(),
             "Chain job finished"
         );
-        Ok(cp)
+        ChainJobOutcome::Finished(cp)
     }
 }
diff --git a/bdk_electrum_streaming/src/lib.rs b/bdk_electrum_streaming/src/lib.rs
index 99ae8df..9f00fa7 100644
--- a/bdk_electrum_streaming/src/lib.rs
+++ b/bdk_electrum_streaming/src/lib.rs
@@ -5,6 +5,8 @@ use bdk_core::spk_client::FullScanResponse;
 pub use electrum_streaming_client;
 
 use bdk_core::ConfirmationBlockTime;
+mod cache;
+pub use cache::*;
 mod state;
 use electrum_streaming_client::{
     AsyncPendingRequest, BlockingPendingRequest, MaybeBatch, PendingRequest,
diff --git a/bdk_electrum_streaming/src/req.rs b/bdk_electrum_streaming/src/req.rs
index b1b70db..54ecce6 100644
--- a/bdk_electrum_streaming/src/req.rs
+++ b/bdk_electrum_streaming/src/req.rs
@@ -87,6 +87,20 @@ impl From<request::HeadersSubscribe> for JobRequest {
     }
 }
 
+/// A response's originating request, alongside what is needed to route it.
+#[derive(Debug, Clone)]
+pub struct PoppedRequest {
+    /// The request this response is answering.
+    pub request: JobRequest,
+    /// Jobs awaiting this response.
+    pub job_ids: BTreeSet<JobId>,
+    /// Whether the local chain dropped blocks after this request was sent.
+    ///
+    /// The server answers from its own chain, so a response which predates a reorg may describe
+    /// a block which is no longer the one we have at that height.
+    pub reorged_since_sent: bool,
+}
+
 /// Request coordinator.
 ///
 /// Associates responses to their requests and requests to their jobs.
@@ -94,10 +108,12 @@ impl From<request::HeadersSubscribe> for JobRequest {
 pub struct ReqCoord {
     /// Next request id.
     next_id: u32,
-    /// Req id -> Req.
-    awaiting_responses: HashMap<u32, JobRequest>,
+    /// Req id -> the request, and the chain generation it was enqueued at.
+    awaiting_responses: HashMap<u32, (JobRequest, u64)>,
     /// So we won't have duplicate requests.
     req_to_job: HashMap<JobRequest, BTreeSet<JobId>>,
+    /// Bumped every time the local chain drops blocks.
+    chain_generation: u64,
 }
 
 impl ReqCoord {
@@ -112,16 +128,40 @@ impl ReqCoord {
         &mut self.next_id
     }
 
-    pub fn pop(&mut self, req_id: u32) -> Option<(JobRequest, BTreeSet<JobId>)> {
-        let any_req = self.awaiting_responses.remove(&req_id)?;
-        let job_ids = self.req_to_job.remove(&any_req).unwrap_or_default();
-        Some((any_req, job_ids))
+    pub fn pop(&mut self, req_id: u32) -> Option<PoppedRequest> {
+        let (request, generation) = self.awaiting_responses.remove(&req_id)?;
+        let job_ids = self.req_to_job.remove(&request).unwrap_or_default();
+        Some(PoppedRequest {
+            request,
+            job_ids,
+            reorged_since_sent: generation < self.chain_generation,
+        })
+    }
+
+    /// Forget every request `job_id` is still waiting on.
+    ///
+    /// A request wanted by another job stays in flight and merely loses `job_id` as an owner.
+    /// One that nothing else wants is dropped outright, so its response is ignored on arrival
+    /// and an identical request is no longer deduplicated against it.
+    pub fn forget_job(&mut self, job_id: JobId) {
+        let mut orphaned = Vec::new();
+        self.req_to_job.retain(|req, job_ids| {
+            if !job_ids.remove(&job_id) || !job_ids.is_empty() {
+                return true;
+            }
+            orphaned.push(req.clone());
+            false
+        });
+        self.awaiting_responses
+            .retain(|_, (req, _)| !orphaned.contains(req));
     }
 
-    /// To be called when the network resets.
-    pub fn clear(&mut self) {
-        self.awaiting_responses.clear();
-        self.req_to_job.clear();
+    /// To be called when the local chain drops blocks.
+    ///
+    /// Requests already in flight were made against the old chain, so their responses can no
+    /// longer be trusted to describe the blocks we now have.
+    pub fn bump_chain_generation(&mut self) {
+        self.chain_generation += 1;
     }
 
     pub fn queuer<'q>(&'q mut self, queue: &'q mut ReqQueue, job_id: JobId) -> ReqQueuer<'q> {
@@ -136,7 +176,7 @@ impl ReqCoord {
     pub fn pending_requests(&self) -> impl ExactSizeIterator<Item = RawRequest> + '_ {
         self.awaiting_responses
             .iter()
-            .map(|(&req_id, req)| req.to_raw(req_id))
+            .map(|(&req_id, (req, _))| req.to_raw(req_id))
     }
 }
 
@@ -163,7 +203,10 @@ impl<'q> ReqQueuer<'q> {
                 e.insert(BTreeSet::new()).insert(self.job_id);
                 let req_id = self.coord.next_id;
                 self.coord.next_id = self.coord.next_id.wrapping_add(1);
-                self.coord.awaiting_responses.insert(req_id, req.clone());
+                let generation = self.coord.chain_generation;
+                self.coord
+                    .awaiting_responses
+                    .insert(req_id, (req.clone(), generation));
                 self.queue.push_back(req.into_raw(req_id));
             }
         }
diff --git a/bdk_electrum_streaming/src/spk_job.rs b/bdk_electrum_streaming/src/spk_job.rs
index af45309..ae7bcc5 100644
--- a/bdk_electrum_streaming/src/spk_job.rs
+++ b/bdk_electrum_streaming/src/spk_job.rs
@@ -19,7 +19,14 @@ pub enum SpkJobStage {
     },
     ProcessingTxsAndAnchors {
         txs: Option<TxsJobStage>,
+        /// The `(height, txid)` pairs to anchor.
+        ///
+        /// Held whole for the life of the job: every pass resolves all of them afresh
+        /// against the chain as it is right then, so a reorg landing mid-job cannot leave the
+        /// job emitting anchors from two different chains.
         anchors: BTreeSet<(u32, Txid)>,
+        /// Whether every anchor resolved on the last pass.
+        anchors_resolved: bool,
     },
 }
 
@@ -28,12 +35,13 @@ impl SpkJobStage {
         Self::ProcessingTxsAndAnchors {
             txs: None,
             anchors: BTreeSet::new(),
+            anchors_resolved: true,
         }
     }
 
     /// Whether it's done.
     pub fn is_done(&self) -> bool {
-        matches!(self, SpkJobStage::ProcessingTxsAndAnchors { txs, anchors } if txs.is_none() && anchors.is_empty())
+        matches!(self, SpkJobStage::ProcessingTxsAndAnchors { txs, anchors_resolved, .. } if txs.is_none() && *anchors_resolved)
     }
 }
 
@@ -107,28 +115,33 @@ impl SpkJob {
     }
 
     pub fn elapsed_seconds(&self) -> String {
-        let duration = UNIX_EPOCH.elapsed().expect("must get current timestamp") - self.start;
-        let seconds = duration.as_secs();
-        let subsec = duration.subsec_millis();
-        format!("{seconds}s {subsec}ms")
+        let now = UNIX_EPOCH.elapsed().expect("must get current timestamp");
+        // The system clock can step backwards, which must not bring a log line down with it.
+        let duration = now.saturating_sub(self.start);
+        format!("{}s {}ms", duration.as_secs(), duration.subsec_millis())
     }
 
     /// Try fullfill all that is missing.
     pub fn advance(mut self, queuer: &mut ReqQueuer, cache: &Cache, cp: &CheckPoint) -> Self {
         let mut made_progress = true;
         while made_progress {
-            (self, made_progress) = self.try_advance_once(queuer, cache, cp.clone());
+            (self, made_progress) = self.try_advance_once(queuer, cache, cp);
             let stage_str = match &self.stage {
                 SpkJobStage::ProcessingHistory { status } => format!("ProcessingHistory({status})"),
-                SpkJobStage::ProcessingTxsAndAnchors { txs, anchors } => {
+                SpkJobStage::ProcessingTxsAndAnchors {
+                    txs,
+                    anchors,
+                    anchors_resolved,
+                } => {
                     let inner_str = match txs {
                         Some(TxsJobStage::Txs(txids)) => format!("txs = {}", txids.len()),
                         Some(TxsJobStage::Prevouts(ops)) => format!("prevouts = {}", ops.len()),
                         None => "tx_done".to_string(),
                     };
                     format!(
-                        "ProcessingTxsAndAnchors({inner_str}, anchors = {})",
-                        anchors.len()
+                        "ProcessingTxsAndAnchors({inner_str}, anchors = {}, resolved = {})",
+                        anchors.len(),
+                        anchors_resolved,
                     )
                 }
             };
@@ -144,17 +157,17 @@ impl SpkJob {
 
     pub fn try_finish(&mut self) -> Option<(ElectrumScriptHash, TxUpdate<ConfirmationBlockTime>)> {
         if self.stage.is_done() {
-            tracing::trace!(
+            tracing::info!(
                 elapsed_seconds = self.elapsed_seconds(),
                 spk_hash = self.spk_hash.to_string(),
-                "Spk job not finished"
+                "Spk job finished"
             );
             Some((self.spk_hash, core::mem::take(&mut self.tx_update)))
         } else {
-            tracing::info!(
+            tracing::trace!(
                 elapsed_seconds = self.elapsed_seconds(),
                 spk_hash = self.spk_hash.to_string(),
-                "Spk job finished"
+                "Spk job not finished"
             );
             None
         }
@@ -167,47 +180,52 @@ impl SpkJob {
         mut self,
         queuer: &mut ReqQueuer,
         cache: &Cache,
-        tip: CheckPoint,
+        tip: &CheckPoint,
     ) -> (Self, bool) {
         match self.stage {
-            SpkJobStage::ProcessingHistory { status } => match cache.spk_histories.get(&status) {
-                Some(history) => {
-                    if let Some(prev_txids) = cache.spk_txids.get(&self.spk_hash) {
-                        let these_txids =
-                            history.iter().map(|tx| tx.txid()).collect::<BTreeSet<_>>();
-                        let to_evict = prev_txids
-                            .difference(&these_txids)
-                            .map(|&txid| (txid, self.start.as_secs()));
-                        self.tx_update.evicted_ats.extend(to_evict);
-                    }
-                    for tx in history {
-                        if let response::Tx::Mempool(tx) = tx {
-                            self.tx_update
-                                .seen_ats
-                                .insert((tx.txid, self.start.as_secs()));
+            SpkJobStage::ProcessingHistory { status } => {
+                match cache.spk_histories.get(self.spk_hash, status) {
+                    Some(history) => {
+                        if let Some(prev_txids) = cache.spk_txids.get(&self.spk_hash) {
+                            let these_txids =
+                                history.iter().map(|tx| tx.txid()).collect::<BTreeSet<_>>();
+                            let to_evict = prev_txids
+                                .difference(&these_txids)
+                                .map(|&txid| (txid, self.start.as_secs()));
+                            self.tx_update.evicted_ats.extend(to_evict);
+                        }
+                        for tx in history {
+                            if let response::Tx::Mempool(tx) = tx {
+                                self.tx_update
+                                    .seen_ats
+                                    .insert((tx.txid, self.start.as_secs()));
+                            }
                         }
-                    }
 
-                    let txs = TxsJobStage::from_missing_txs(history.iter().map(|tx| tx.txid()));
-                    let anchors = history
-                        .iter()
-                        .filter_map(|tx| {
-                            let height = tx.confirmation_height()?.to_consensus_u32();
-                            Some((height, tx.txid()))
-                        })
-                        .collect();
-                    self.stage = SpkJobStage::ProcessingTxsAndAnchors { txs, anchors };
-                    (self, true)
-                }
-                None => {
-                    let script_hash = self.spk_hash;
-                    queuer.enqueue(request::GetHistory { script_hash });
-                    (self, false)
+                        let txs = TxsJobStage::from_missing_txs(history.iter().map(|tx| tx.txid()));
+                        let anchors = history
+                            .iter()
+                            .filter_map(|tx| {
+                                let height = tx.confirmation_height()?.to_consensus_u32();
+                                Some((height, tx.txid()))
+                            })
+                            .collect();
+                        self.stage = SpkJobStage::ProcessingTxsAndAnchors {
+                            txs,
+                            anchors,
+                            anchors_resolved: false,
+                        };
+                        (self, true)
+                    }
+                    None => {
+                        let script_hash = self.spk_hash;
+                        queuer.enqueue(request::GetHistory { script_hash });
+                        (self, false)
+                    }
                 }
-            },
+            }
             SpkJobStage::ProcessingTxsAndAnchors {
-                mut txs,
-                mut anchors,
+                mut txs, anchors, ..
             } => {
                 let mut made_progress = false;
                 txs = match txs {
@@ -266,44 +284,74 @@ impl SpkJob {
                     None => None,
                 };
 
-                let anchors_start_count = anchors.len();
-                anchors.retain(|&(height, txid)| {
-                    if height > tip.height() {
-                        // Nothing to request for a block we don't know exists yet. The job is
-                        // re-advanced once a chain job advances the tip.
-                        return true;
-                    }
-
-                    let blockhash = match tip.get(height) {
-                        Some(cp) if cp.height() == height => cp.hash(),
-                        _ => {
-                            queuer.enqueue(request::Header { height });
-                            return true;
-                        }
-                    };
+                // Anchors are resolved from scratch each pass, so this never leaves a
+                // partially resolved set staged in the update.
+                let resolved = advance_anchors(queuer, cache, tip, &anchors);
+                let anchors_resolved = resolved.is_some();
+                self.tx_update.anchors = resolved.unwrap_or_default();
 
-                    if !cache.headers.contains_key(&blockhash) {
-                        queuer.enqueue(request::Header { height });
-                    }
+                self.stage = SpkJobStage::ProcessingTxsAndAnchors {
+                    txs,
+                    anchors,
+                    anchors_resolved,
+                };
+                (self, made_progress)
+            }
+        }
+    }
+}
 
-                    if let Some(anchor) = cache.anchors.get(&(txid, blockhash)) {
-                        self.tx_update.anchors.insert((*anchor, txid));
-                        return false;
-                    };
-                    if cache.failed_anchors.contains(&(txid, blockhash)) {
-                        return false;
-                    }
+/// Resolve `anchors` against `cache`, queueing whatever is missing.
+///
+/// Returns the resolved anchors, but only once every one of them resolved — nothing is held on to
+/// between calls. Anchors are therefore always resolved as a set against a single chain: a reorg
+/// landing midway through simply has the next call resolve all of them against the chain we moved
+/// to, instead of leaving a job to emit anchors from the chain it started on next to anchors from
+/// the chain it ended on.
+///
+/// This is also why anchors are tracked as `(height, txid)` pairs rather than by block hash: each
+/// call resolves them against whichever block `tip` currently has at that height.
+fn advance_anchors(
+    queuer: &mut ReqQueuer,
+    cache: &Cache,
+    tip: &CheckPoint,
+    anchors: &BTreeSet<(u32, Txid)>,
+) -> Option<BTreeSet<(ConfirmationBlockTime, Txid)>> {
+    let mut resolved = BTreeSet::new();
+    let mut all_resolved = true;
 
-                    queuer.enqueue(request::GetTxMerkle { txid, height });
-                    true
-                });
-                if anchors.len() < anchors_start_count {
-                    made_progress = true;
-                }
+    for &(height, txid) in anchors {
+        if height > tip.height() {
+            // Nothing to request for a block we don't know exists yet. The job is
+            // re-advanced once a chain job advances the tip.
+            all_resolved = false;
+            continue;
+        }
 
-                self.stage = SpkJobStage::ProcessingTxsAndAnchors { txs, anchors };
-                (self, made_progress)
+        let blockhash = match tip.get(height) {
+            Some(cp) => cp.hash(),
+            None => {
+                queuer.enqueue(request::Header { height });
+                all_resolved = false;
+                continue;
             }
+        };
+
+        if let Some(anchor) = cache.anchors.get(&(txid, blockhash)) {
+            resolved.insert((*anchor, txid));
+            continue;
         }
+        // A proof is verified against this block's merkle root, so asking for the proof before
+        // the header is cached would make the outcome depend on the server answering in request
+        // order — which the protocol's request ids exist precisely because it does not promise.
+        if !cache.headers.contains_key(&blockhash) {
+            queuer.enqueue(request::Header { height });
+            all_resolved = false;
+            continue;
+        }
+
+        queuer.enqueue(request::GetTxMerkle { txid, height });
+        all_resolved = false;
     }
+    all_resolved.then_some(resolved)
 }
diff --git a/bdk_electrum_streaming/src/state.rs b/bdk_electrum_streaming/src/state.rs
index 49d3e2e..e6e57af 100644
--- a/bdk_electrum_streaming/src/state.rs
+++ b/bdk_electrum_streaming/src/state.rs
@@ -1,15 +1,9 @@
-use std::{
-    collections::{BTreeMap, BTreeSet, HashMap, HashSet},
-    sync::Arc,
-};
+use std::collections::{btree_map, BTreeMap, BTreeSet};
 
 use anyhow::Context;
-use bdk_core::{
-    bitcoin::{self, BlockHash, Transaction, Txid},
-    BlockId, CheckPoint, ConfirmationBlockTime,
-};
+use bdk_core::{BlockId, CheckPoint, ConfirmationBlockTime};
 use electrum_streaming_client::{
-    notification::Notification, request, response, AsyncPendingRequest, BlockingPendingRequest,
+    notification::Notification, request, AsyncPendingRequest, BlockingPendingRequest,
     ElectrumScriptHash, ElectrumScriptStatus, MaybeBatch, PendingRequest,
     RawNotificationOrResponse, Request,
 };
@@ -17,8 +11,9 @@ use miniscript::{Descriptor, DescriptorPublicKey};
 use serde_json::from_value;
 
 use crate::{
-    chain_job::ChainJob,
-    req::{JobRequest, ReqCoord, ReqQueue},
+    cache::{Cache, SpkHistories},
+    chain_job::{ChainJob, ChainJobOutcome},
+    req::{JobRequest, PoppedRequest, ReqCoord, ReqQueue},
     spk_job::SpkJob,
     DerivedSpkTracker, Update,
 };
@@ -82,6 +77,10 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
         &self.cache
     }
 
+    pub fn spk_histories(&self) -> &SpkHistories {
+        &self.cache.spk_histories
+    }
+
     /// Reset the state to be not initialized.
     ///
     /// Call this after disconnection otherwise pending requests will not be resent and no
@@ -161,6 +160,20 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
                     .context("Failed to deserialize notification from server")?;
                 match notification {
                     Notification::Header(header_notification) => {
+                        // A same-height reorg is applied by `ChainJob`'s short-circuit without
+                        // fetching anything, so this notification is the only place the
+                        // replacement header is ever offered to us. Caching it here saves the
+                        // anchor refetch a round-trip on the very path it exists for.
+                        let header = *header_notification.header();
+                        self.cache.headers.insert(header.block_hash(), header);
+
+                        // A new tip supersedes whatever the previous job was resolving against,
+                        // and its request has to go with it. The replacement job asks for the
+                        // same heights, so a surviving request would be deduplicated against —
+                        // nothing would be sent, and the answer already in flight would fill the
+                        // new job with the chain the server has just left.
+                        self.coord.forget_job(JobId::Chain);
+
                         // Always replace prev job since a new notification means a new tip.
                         self.chain_job = ChainJob::new(
                             self.coord.queuer(req_queue, JobId::Chain),
@@ -168,17 +181,7 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
                             *header_notification.header(),
                             header_notification.height(),
                         );
-                        if let Some(job) = self.chain_job.take() {
-                            match job.try_finish(&mut self.cp) {
-                                Ok(cp) => Ok(Some(self.on_chain_job_completed(req_queue, cp))),
-                                Err(job) => {
-                                    self.chain_job = Some(job);
-                                    Ok(None)
-                                }
-                            }
-                        } else {
-                            Ok(None)
-                        }
+                        Ok(self.try_finish_chain_job(req_queue))
                     }
                     Notification::ScriptHash(script_hash_notification) => {
                         let spk_hash = script_hash_notification.script_hash();
@@ -194,6 +197,10 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
 
                         let mut last_active_indices = BTreeMap::new();
 
+                        if spk_status.is_none() {
+                            self.cache.spk_histories.remove(spk_hash);
+                        }
+
                         if spk_status.is_some() || self.cache.spk_txids.contains_key(&spk_hash) {
                             for script_hash in self.spk_tracker.mark_script_hash_used(&k, i) {
                                 self.coord
@@ -227,7 +234,11 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
                 }
             }
             RawNotificationOrResponse::Response(raw_response) => {
-                let (orig_req, job_ids) = match self.coord.pop(raw_response.id) {
+                let PoppedRequest {
+                    request: orig_req,
+                    job_ids,
+                    reorged_since_sent,
+                } = match self.coord.pop(raw_response.id) {
                     Some(req) => req,
                     None => return Ok(None),
                 };
@@ -236,6 +247,38 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
                 let raw = match raw_response.result {
                     Ok(raw) => raw,
                     Err(err) => {
+                        // An anchor fetch is speculative: it asks for a proof of inclusion at
+                        // the height a transaction was last reported at, and a reorg may have
+                        // taken the transaction out of that block, or out of the chain
+                        // altogether. A server answers that with an error, which is an answer,
+                        // not a reason to bring the connection down.
+                        if let JobRequest::GetTxMerkle(req) = &orig_req {
+                            tracing::warn!(
+                                txid = req.txid.to_string(),
+                                block_height = req.height,
+                                ?err,
+                                "Server gave no merkle proof at this height. Reorg?",
+                            );
+                            // An error about the chain we have since left says nothing about
+                            // the block we now have at this height. Discard it and let the
+                            // jobs ask again.
+                            if reorged_since_sent {
+                                return Ok(self.advance_spk_jobs(req_queue, job_ids));
+                            }
+                            // An error proves nothing — it is equally a rate limit, an index
+                            // still catching up or a daemon hiccup — so nothing durable is
+                            // recorded, and there is nowhere to record it: a proof
+                            // request that comes back with nothing leaves no trace.
+                            //
+                            // The job is left stashed rather than cancelled. Returning without
+                            // advancing it is what stops the re-ask loop; keeping it is what
+                            // makes the recovery automatic, since the next chain update advances
+                            // it again and the answer may be different once our chain has caught
+                            // up with the server's. Cancelling would need a notification to
+                            // revive it, and the same-height reorg this all exists for is
+                            // precisely the case that sends none.
+                            return Ok(None);
+                        }
                         // Cancel jobs that resulted in error.
                         self.cancel_jobs(job_ids);
                         return Err(anyhow::anyhow!(err).context("Server responded with error"));
@@ -252,55 +295,65 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
                         if let Some(job) = self.chain_job.take() {
                             let new_blocks = (req.start_height..)
                                 .zip(resp.headers.into_iter().map(|h| h.block_hash()));
-                            match job.process_blocks(new_blocks).try_finish(&mut self.cp) {
-                                Ok(cp) => Ok(Some(self.on_chain_job_completed(req_queue, cp))),
-                                Err(job) => {
-                                    self.chain_job = Some(job);
-                                    Ok(None)
-                                }
-                            }
-                        } else {
-                            Ok(None)
+                            self.chain_job = Some(job.process_blocks(new_blocks));
                         }
+                        Ok(self.try_finish_chain_job(req_queue))
                     }
                     JobRequest::GetHeader(req) => {
                         let resp = from_raw(&req, raw)?;
+                        let hash = resp.header.block_hash();
+                        self.cache.headers.insert(hash, resp.header);
 
-                        self.cache
-                            .headers
-                            .insert(resp.header.block_hash(), resp.header);
-
-                        // Do not extend checkpoints.
-                        if req.height > self.cp.height() {
-                            return Ok(None);
-                        }
-                        // Do not replace blocks.
-                        if self
+                        // The block our own chain has at this height, if it has one.
+                        let ours = self
                             .cp
                             .get(req.height)
-                            .is_some_and(|cp| cp.height() == req.height)
-                        {
+                            .filter(|cp| cp.height() == req.height)
+                            .map(|cp| cp.hash());
+
+                        // `blockchain.block.header` is keyed by height, so this is the only
+                        // block the server will ever hand us there. If our chain claims a
+                        // different one it is stale at a height no chain job will rewrite —
+                        // below `ChainJob`'s suffix nothing reports the difference — and the
+                        // header a job is waiting on can never arrive. Asking again would loop
+                        // for as long as the connection is up, so drop the jobs instead and
+                        // let the next notification rebuild them.
+                        if ours.is_some_and(|ours| ours != hash) && !reorged_since_sent {
+                            tracing::warn!(
+                                height = req.height,
+                                ours = ours.expect("just matched").to_string(),
+                                theirs = hash.to_string(),
+                                "Local chain disagrees with the server below the reorg horizon",
+                            );
+                            self.cancel_jobs(job_ids);
                             return Ok(None);
                         }
-                        self.cp = self
-                            .cp
-                            .clone()
-                            .insert(BlockId::from((req.height, resp.header.block_hash())));
+
+                        // Whether or not the checkpoint chain wants this header, the header
+                        // now being cached is what a waiting anchor may need, and nothing else
+                        // will wake it. So only the insert below is conditional; the jobs are
+                        // advanced either way.
+                        //
+                        // The insert is skipped when the block at this height may have been
+                        // replaced since we asked (`reorged_since_sent`), when it would extend
+                        // the checkpoints, and when we already have this block.
+                        let extends = req.height > self.cp.height();
+                        if !reorged_since_sent && !extends && ours.is_none() {
+                            self.cp = self.cp.clone().insert(BlockId::from((req.height, hash)));
+                        }
                         Ok(self.advance_spk_jobs(req_queue, job_ids))
                     }
                     JobRequest::GetHistory(req) => {
                         let resp = from_raw(&req, raw)?;
                         if let Some(spk_status) = ElectrumScriptStatus::from_history(&resp) {
-                            self.cache
-                                .spk_histories
-                                .entry(spk_status)
-                                .or_default()
-                                .extend(resp.clone());
                             self.cache
                                 .spk_txids
                                 .entry(req.script_hash)
                                 .or_default()
                                 .extend(resp.iter().map(|tx| tx.txid()));
+                            self.cache
+                                .spk_histories
+                                .insert(req.script_hash, spk_status, resp);
                         }
                         Ok(self.advance_spk_jobs(req_queue, job_ids))
                     }
@@ -311,6 +364,15 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
                     }
                     JobRequest::GetTxMerkle(req) => {
                         let resp = from_raw(&req, raw)?;
+
+                        // The proof was built against the server's chain at the time we asked,
+                        // which may not be the block we now have at this height. Checking it
+                        // against that block would read a disagreement between two chains as a
+                        // verdict on this one, so discard it and let the job ask again.
+                        if reorged_since_sent {
+                            return Ok(self.advance_spk_jobs(req_queue, job_ids));
+                        }
+
                         let cp = match self.cp.get(req.height) {
                             Some(cp) if cp.height() == req.height => cp,
                             _ => {
@@ -324,7 +386,7 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
                             }
                         };
                         let header = match self.cache.headers.get(&cp.hash()) {
-                            Some(header) => header,
+                            Some(header) => *header,
                             None => {
                                 tracing::warn!(
                                     ?req,
@@ -357,11 +419,16 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
                                 block_hash = header.block_hash().to_string(),
                                 header_root = header.merkle_root.to_string(),
                                 expected_root = exp_root.to_string(),
-                                "Failed to verify anchor."
+                                "Proof does not match the block we have at this height",
                             );
-                            self.cache
-                                .failed_anchors
-                                .insert((req.txid, header.block_hash()));
+                            // The server proves inclusion in whichever block *it* has at this
+                            // height, so a mismatch says our chain and the server's disagree
+                            // there — not that the transaction is absent from our block. That is
+                            // the conclusion `GetHeader` draws from the same kind of evidence,
+                            // so it gets the same treatment: drop the jobs and let the next
+                            // notification rebuild them.
+                            self.cancel_jobs(job_ids);
+                            return Ok(None);
                         }
                         Ok(self.advance_spk_jobs(req_queue, job_ids))
                     }
@@ -379,6 +446,10 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
 
                         let mut last_active_indices = BTreeMap::new();
 
+                        if spk_status.is_none() {
+                            self.cache.spk_histories.remove(spk_hash);
+                        }
+
                         if spk_status.is_some() || self.cache.spk_txids.contains_key(&spk_hash) {
                             for script_hash in self.spk_tracker.mark_script_hash_used(&k, i) {
                                 self.coord
@@ -408,6 +479,9 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
                     }
                     JobRequest::HeadersSubscribe(req) => {
                         let resp = from_raw(&req, raw)?;
+                        self.cache
+                            .headers
+                            .insert(resp.header.block_hash(), resp.header);
 
                         // Always replace prev job since a new notification means a new tip.
                         self.chain_job = ChainJob::new(
@@ -416,27 +490,78 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
                             resp.header,
                             resp.height,
                         );
-                        if let Some(job) = self.chain_job.take() {
-                            match job.try_finish(&mut self.cp) {
-                                Ok(cp) => Ok(Some(self.on_chain_job_completed(req_queue, cp))),
-                                Err(job) => {
-                                    self.chain_job = Some(job);
-                                    Ok(None)
-                                }
-                            }
-                        } else {
-                            Ok(None)
-                        }
+                        Ok(self.try_finish_chain_job(req_queue))
                     }
                 }
             }
         }
     }
 
+    /// Apply the pending chain job to the local chain, if it has everything it needs.
+    ///
+    /// Returns the resulting update, if the job completed.
+    fn try_finish_chain_job(&mut self, req_queue: &mut ReqQueue) -> Option<Update<K>> {
+        let job = self.chain_job.take()?;
+        let prev_cp = self.cp.clone();
+        match job.try_finish(&mut self.cp) {
+            ChainJobOutcome::Finished(cp) => {
+                Some(self.on_chain_job_completed(req_queue, &prev_cp, cp))
+            }
+            ChainJobOutcome::Pending(job) => {
+                self.chain_job = Some(job);
+                None
+            }
+            ChainJobOutcome::Superseded => None,
+        }
+    }
+
+    /// React to the local chain having moved from `prev_cp` to `cp`.
+    ///
     /// Spk jobs cannot extend the local chain, so a job whose anchor is above the local tip
     /// waits with no request in flight. Advancing the tip is what makes such an anchor
     /// resolvable, so every completed chain job must re-advance the stashed spk jobs.
-    fn on_chain_job_completed(&mut self, req_queue: &mut ReqQueue, cp: CheckPoint) -> Update<K> {
+    ///
+    /// A chain job may also drop blocks. Any transaction we have seen at an evicted height is
+    /// anchored to a block which is no longer ours, so its anchor has to be refetched — and a
+    /// spk notification will not tell us to, since a transaction which moved to a different
+    /// block of the same height leaves the spk status untouched.
+    fn on_chain_job_completed(
+        &mut self,
+        req_queue: &mut ReqQueue,
+        prev_cp: &CheckPoint,
+        cp: CheckPoint,
+    ) -> Update<K> {
+        let evicted = evicted_heights(prev_cp, &cp);
+        if !evicted.is_empty() {
+            tracing::info!(
+                heights = ?evicted,
+                "Blocks evicted from the local chain. Refetching anchors."
+            );
+            // Responses to requests which are still in flight describe the chain we just left
+            // behind.
+            self.coord.bump_chain_generation();
+
+            // Start each affected script the job its own notification would have started, from
+            // the status and history already cached: a script hash notification we raise
+            // ourselves, because the server will not raise one. A transaction which moved to a
+            // different block of the same height leaves the status untouched.
+            for spk_hash in self.cache.spk_histories.spk_hashes_at_heights(evicted) {
+                // A vacant entry is the whole rule: never displace a job the server's own
+                // notification built, since that one carries a status at least as new as
+                // anything we could replay, and every stashed job is re-advanced below anyway.
+                if let btree_map::Entry::Vacant(e) = self.spk_jobs.entry(spk_hash) {
+                    if let Some(status) = self.cache.spk_histories.status(spk_hash) {
+                        e.insert(SpkJob::new(&self.cache, spk_hash, Some(status)));
+                    }
+                }
+            }
+        }
+
+        // Only heights inside `ChainJob`'s reorg window can ever be evicted, and only evicted
+        // heights are ever read back, so entries far below the tip can never be consulted
+        // again. Pruning keeps this bounded with a wide margin over that horizon.
+        self.cache.spk_histories.prune(cp.height());
+
         let stashed_jobs = self
             .spk_jobs
             .keys()
@@ -497,20 +622,26 @@ impl<PReq: PendingRequest, K: Ord + Clone> State<PReq, K> {
     }
 }
 
+/// The heights whose block was dropped from the local chain when it went from `prev` to `next`.
+///
+/// Checkpoint chains share everything below the point at which they diverge, so walking `prev`
+/// down from its tip until `next` agrees is enough.
+fn evicted_heights(prev: &CheckPoint, next: &CheckPoint) -> BTreeSet<u32> {
+    let mut evicted = BTreeSet::new();
+    for cp in prev.iter() {
+        match next.get(cp.height()) {
+            Some(next_cp) if next_cp.hash() == cp.hash() => break,
+            _ => {
+                evicted.insert(cp.height());
+            }
+        }
+    }
+    evicted
+}
+
 pub fn from_raw<R>(_req: &R, raw: serde_json::Value) -> Result<R::Response, serde_json::Error>
 where
     R: Request,
 {
     from_value(raw)
 }
-
-/// A monotonically growing cache.
-#[derive(Debug, Clone, Default)]
-pub struct Cache {
-    pub spk_histories: HashMap<ElectrumScriptStatus, Vec<response::Tx>>,
-    pub spk_txids: HashMap<ElectrumScriptHash, BTreeSet<Txid>>,
-    pub txs: HashMap<Txid, Arc<Transaction>>,
-    pub anchors: HashMap<(Txid, BlockHash), ConfirmationBlockTime>,
-    pub failed_anchors: HashSet<(Txid, BlockHash)>,
-    pub headers: HashMap<BlockHash, bitcoin::block::Header>,
-}
diff --git a/bdk_electrum_streaming/tests/env.rs b/bdk_electrum_streaming/tests/env.rs
index 1662d9f..3bedfa7 100644
--- a/bdk_electrum_streaming/tests/env.rs
+++ b/bdk_electrum_streaming/tests/env.rs
@@ -5,7 +5,7 @@ use bdk_chain::{
     ChainPosition, IndexedTxGraph,
 };
 use bdk_core::{
-    bitcoin::{key::Secp256k1, params::REGTEST, Address, Amount},
+    bitcoin::{key::Secp256k1, params::REGTEST, Address, Amount, BlockHash, Txid},
     ConfirmationBlockTime,
 };
 use bdk_electrum_streaming::{
@@ -359,3 +359,242 @@ async fn new_block_confirmation_is_anchored_live() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+type Graph = IndexedTxGraph<ConfirmationBlockTime, KeychainTxOutIndex<&'static str>>;
+
+/// The anchor `txid` is canonically confirmed at, if it is confirmed at all.
+fn canonical_anchor(
+    chain: &LocalChain,
+    graph: &Graph,
+    txid: Txid,
+) -> Option<ConfirmationBlockTime> {
+    graph
+        .graph()
+        .list_canonical_txs(
+            chain,
+            chain.tip().block_id(),
+            CanonicalizationParams::default(),
+        )
+        .find(|ctx| ctx.tx_node.txid == txid)
+        .and_then(|ctx| match ctx.chain_position {
+            ChainPosition::Confirmed { anchor, .. } => Some(anchor),
+            ChainPosition::Unconfirmed { .. } => None,
+        })
+}
+
+/// A live client against a fresh `electrsd`, with the machinery the reorg tests share.
+struct LiveWallet {
+    env: TestEnv,
+    chain: LocalChain,
+    graph: Graph,
+    update_rx: mpsc::UnboundedReceiver<Update<&'static str>>,
+    client: AsyncClient<&'static str>,
+    run_handle: tokio::task::JoinHandle<anyhow::Result<()>>,
+}
+
+impl LiveWallet {
+    /// Connect to a fresh test environment and apply the first (genesis) update.
+    async fn new() -> anyhow::Result<Self> {
+        init();
+
+        let secp = Secp256k1::new();
+        let env = TestEnv::new()?;
+        let electrum_url = env.electrsd.electrum_url.clone();
+
+        let (external, _) = Descriptor::parse_descriptor(&secp, DESCRIPTORS[0])?;
+        let (internal, _) = Descriptor::parse_descriptor(&secp, DESCRIPTORS[1])?;
+
+        let mut graph = IndexedTxGraph::<ConfirmationBlockTime, _>::new({
+            let mut indexer = KeychainTxOutIndex::<&'static str>::new(LOOKAHEAD, false);
+            indexer.insert_descriptor(EXTERNAL, external.clone())?;
+            indexer.insert_descriptor(INTERNAL, internal.clone())?;
+            indexer
+        });
+        let (mut chain, _cs) = LocalChain::from_genesis_hash(env.genesis_hash()?);
+
+        let mut spk_tracker = DerivedSpkTracker::<&'static str>::new(LOOKAHEAD);
+        spk_tracker.insert_descriptor(EXTERNAL, external, 0);
+        spk_tracker.insert_descriptor(INTERNAL, internal, 0);
+
+        let mut state = AsyncState::new(
+            ReqCoord::default(),
+            Cache::default(),
+            spk_tracker,
+            chain.tip(),
+        );
+
+        let (mut update_tx, mut update_rx) = mpsc::unbounded::<Update<&'static str>>();
+        let (client, mut client_rx) = AsyncClient::new();
+
+        let run_handle = tokio::spawn(async move {
+            let mut conn = TcpStream::connect(&electrum_url).await?;
+            let (read, write) = conn.split();
+            run_async(
+                &mut state,
+                &mut update_tx,
+                &mut client_rx,
+                read.compat(),
+                write.compat_write(),
+            )
+            .await?;
+            anyhow::Ok(())
+        });
+
+        let update = update_rx.next().await.expect("Must have next update");
+        apply_update(&mut chain, &mut graph, update)?;
+
+        Ok(Self {
+            env,
+            chain,
+            graph,
+            update_rx,
+            client,
+            run_handle,
+        })
+    }
+
+    /// Apply updates until `f` holds.
+    ///
+    /// Errors if the client stops — which is what a connection torn down by an expected server
+    /// error looks like from here — or if `f` has not held within the timeout.
+    async fn wait_until(
+        &mut self,
+        what: &str,
+        mut f: impl FnMut(&LocalChain, &Graph) -> bool,
+    ) -> anyhow::Result<()> {
+        let timeout = tokio::time::sleep(Duration::from_secs(150)).fuse();
+        pin_mut!(timeout);
+        loop {
+            if f(&self.chain, &self.graph) {
+                return Ok(());
+            }
+            futures::select! {
+                _ = timeout => return Err(anyhow::anyhow!("timed out waiting for {what}")),
+                update = self.update_rx.next() => {
+                    let update = update.ok_or_else(|| {
+                        anyhow::anyhow!("the client stopped while waiting for {what}")
+                    })?;
+                    apply_update(&mut self.chain, &mut self.graph, update)?;
+                },
+            }
+        }
+    }
+
+    /// Mine past coinbase maturity, then send `Amount::ONE_BTC` to a tracked spk and mine it in.
+    ///
+    /// Returns the txid and the hash of the block confirming it.
+    async fn confirm_tracked_tx(&mut self) -> anyhow::Result<(Txid, BlockHash)> {
+        self.env.mine_blocks(101, None)?;
+        let premine_height = self.env.rpc_client().get_block_count()? as u32;
+        self.wait_until("the premined chain", |chain, _| {
+            chain.tip().height() >= premine_height
+        })
+        .await?;
+
+        let ((_, spk), _) = self
+            .graph
+            .index
+            .next_unused_spk(EXTERNAL)
+            .expect("must derive spk");
+        let txid = self
+            .env
+            .send(&Address::from_script(&spk, &REGTEST)?, Amount::ONE_BTC)?;
+        self.wait_until("the unconfirmed tx", |_, graph| {
+            graph.graph().get_tx(txid).is_some()
+        })
+        .await?;
+
+        self.env.mine_blocks(1, None)?;
+        self.wait_until("the tx to be anchored", |chain, graph| {
+            canonical_anchor(chain, graph, txid).is_some()
+        })
+        .await?;
+
+        let anchor = canonical_anchor(&self.chain, &self.graph, txid).expect("just waited for it");
+        Ok((txid, anchor.block_id.hash))
+    }
+
+    async fn stop(self) -> anyhow::Result<()> {
+        self.client.stop().await?;
+        self.run_handle.await??;
+        Ok(())
+    }
+}
+
+/// Issue #12, end to end: a reorg re-mines a confirmed tx into a *different* block at the *same*
+/// height. The Electrum script status is a hash over txid-height pairs, so it is unchanged and no
+/// script hash notification is sent. The anchor must still be refetched off the tip alone.
+#[tokio::test]
+async fn reorg_to_same_height_block_refetches_anchor_live() -> anyhow::Result<()> {
+    let mut w = LiveWallet::new().await?;
+    let (txid, first_block) = w.confirm_tracked_tx().await?;
+    let confirm_height = w.env.rpc_client().get_block_count()? as u32;
+
+    // Invalidate the confirming block and re-mine at the same height. The tx is back in the
+    // mempool, so it goes into the replacement block too.
+    w.env.reorg(1)?;
+    let second_block = w.env.rpc_client().get_best_block_hash()?;
+    assert_ne!(
+        first_block, second_block,
+        "the reorg must actually replace the block"
+    );
+    assert_eq!(
+        w.env.rpc_client().get_block_count()? as u32,
+        confirm_height,
+        "the replacement block must be at the same height"
+    );
+    assert!(
+        w.env
+            .rpc_client()
+            .get_block(&second_block)?
+            .txdata
+            .iter()
+            .any(|tx| tx.compute_txid() == txid),
+        "the replacement block must still contain the tx"
+    );
+
+    w.wait_until("the refetched anchor", |chain, graph| {
+        canonical_anchor(chain, graph, txid).is_some_and(|a| a.block_id.hash == second_block)
+    })
+    .await?;
+
+    let anchor = canonical_anchor(&w.chain, &w.graph, txid).expect("just waited for it");
+    assert_eq!(anchor.block_id.height, confirm_height);
+
+    w.stop().await
+}
+
+/// The everyday reorg: one which takes a tx out of its block and back to the mempool. The anchor
+/// refetch then asks for a proof the server cannot give, and that error must not take the
+/// connection down with it.
+#[tokio::test]
+async fn reorg_unconfirming_a_tx_keeps_the_connection_alive() -> anyhow::Result<()> {
+    let mut w = LiveWallet::new().await?;
+    let (txid, _) = w.confirm_tracked_tx().await?;
+    let confirm_height = w.env.rpc_client().get_block_count()? as u32;
+
+    // Invalidate the confirming block and replace it with empty ones, so the tx cannot be
+    // re-mined and the server has no proof to give at that height.
+    w.env.invalidate_blocks(1)?;
+    w.env.mine_empty_block()?;
+    w.env.mine_empty_block()?;
+    let tip_height = w.env.rpc_client().get_block_count()? as u32;
+    assert_eq!(tip_height, confirm_height + 1);
+    assert!(
+        w.env.rpc_client().get_raw_mempool()?.contains(&txid),
+        "the tx must be back in the mempool, so the server really has no proof for it"
+    );
+
+    // The connection has to keep serving: `wait_until` fails if the client stops.
+    w.wait_until("the chain tip after the reorg", |chain, _| {
+        chain.tip().height() >= tip_height
+    })
+    .await?;
+
+    assert!(
+        canonical_anchor(&w.chain, &w.graph, txid).is_none(),
+        "the tx must no longer be canonically confirmed"
+    );
+
+    w.stop().await
+}
diff --git a/bdk_electrum_streaming/tests/state.rs b/bdk_electrum_streaming/tests/state.rs
index f95fb27..934eab9 100644
--- a/bdk_electrum_streaming/tests/state.rs
+++ b/bdk_electrum_streaming/tests/state.rs
@@ -2,9 +2,12 @@ use std::{str::FromStr, sync::Arc};
 
 use bdk_core::{
     bitcoin::{
-        absolute, block, consensus::encode::serialize_hex, constants, hashes::Hash, transaction,
-        Amount, CompactTarget, Network, OutPoint, ScriptBuf, Sequence, Transaction, TxIn,
-        TxMerkleNode, TxOut, Txid, Witness,
+        absolute, block,
+        consensus::encode::serialize_hex,
+        constants,
+        hashes::{sha256d, Hash},
+        transaction, Amount, CompactTarget, Network, OutPoint, ScriptBuf, Sequence, Transaction,
+        TxIn, TxMerkleNode, TxOut, Txid, Witness,
     },
     BlockId, CheckPoint, ConfirmationBlockTime,
 };
@@ -23,10 +26,14 @@ fn raw_msg(v: serde_json::Value) -> RawNotificationOrResponse {
     serde_json::from_value(v).expect("must deserialize raw message")
 }
 
+#[derive(Clone)]
 struct Server {
     headers: Vec<block::Header>,
     spk_hash: ElectrumScriptHash,
-    tx: Transaction,
+    /// The transactions in `spk_hash`'s history, and the height each is confirmed at.
+    txs: Vec<(Transaction, u32)>,
+    /// The merkle branch and position this server answers every merkle request with.
+    merkle_proof: (Vec<sha256d::Hash>, usize),
 }
 
 impl Server {
@@ -34,8 +41,27 @@ impl Server {
         self.headers.len() - 1
     }
 
-    fn answer(&self, req: &RawRequest) -> serde_json::Value {
-        match req.method.as_ref() {
+    /// The history of `spk_hash`: every tx the chain is long enough to contain.
+    fn history(&self, spk_hash: &serde_json::Value) -> Vec<response::Tx> {
+        if *spk_hash != json!(self.spk_hash.to_string()) {
+            return Vec::new();
+        }
+        self.txs
+            .iter()
+            .filter(|(_, height)| self.tip_height() >= *height as usize)
+            .map(|(tx, height)| {
+                response::Tx::Confirmed(response::ConfirmedTx {
+                    txid: tx.compute_txid(),
+                    height: absolute::Height::from_consensus(*height)
+                        .expect("must be a valid height"),
+                })
+            })
+            .collect()
+    }
+
+    /// Answer a request, or fail it the way a server would.
+    fn answer(&self, req: &RawRequest) -> Result<serde_json::Value, String> {
+        Ok(match req.method.as_ref() {
             "blockchain.headers.subscribe" => {
                 let tip = self.headers.last().expect("server must have blocks");
                 json!({ "hex": serialize_hex(tip), "height": self.tip_height() })
@@ -49,20 +75,58 @@ impl Server {
                     .collect::<String>();
                 json!({ "count": count, "hex": hex, "max": 2016 })
             }
-            "blockchain.scripthash.subscribe" => json!(null),
-            "blockchain.scripthash.get_history" => {
-                if req.params[0] == json!(self.spk_hash.to_string()) && self.tip_height() >= 2 {
-                    json!([{ "tx_hash": self.tx.compute_txid().to_string(), "height": 2 }])
-                } else {
-                    json!([])
+            "blockchain.block.header" => {
+                let height = req.params[0].as_u64().expect("must have height") as usize;
+                match self.headers.get(height) {
+                    Some(header) => json!(serialize_hex(header)),
+                    None => return Err(format!("height {height} is above the chain tip")),
                 }
             }
-            "blockchain.transaction.get" => json!(serialize_hex(&self.tx)),
+            "blockchain.scripthash.subscribe" => {
+                match ElectrumScriptStatus::from_history(&self.history(&req.params[0])) {
+                    Some(status) => json!(status.to_string()),
+                    None => json!(null),
+                }
+            }
+            "blockchain.scripthash.get_history" => json!(self
+                .history(&req.params[0])
+                .iter()
+                .map(|tx| json!({
+                    "tx_hash": tx.txid().to_string(),
+                    "height": tx.electrum_height(),
+                }))
+                .collect::<Vec<_>>()),
+            "blockchain.transaction.get" => {
+                let (tx, _) = self
+                    .txs
+                    .iter()
+                    .find(|(tx, _)| req.params[0] == json!(tx.compute_txid().to_string()))
+                    .expect("must be a tx the server knows");
+                json!(serialize_hex(tx))
+            }
             "blockchain.transaction.get_merkle" => {
-                json!({ "block_height": req.params[1], "merkle": [], "pos": 0 })
+                // A proof can only be given for a tx the server has in that block. Both
+                // romanz/electrs and ElectrumX raise an error otherwise.
+                let height = req.params[1].as_u64().expect("must have height") as u32;
+                if !self
+                    .txs
+                    .iter()
+                    .any(|(tx, h)| *h == height && req.params[0] == json!(tx.compute_txid()))
+                {
+                    return Err(format!(
+                        "tx {} not in block at height {height}",
+                        req.params[0]
+                    ));
+                }
+                let (branch, pos) = &self.merkle_proof;
+                json!({
+                    "block_height": req.params[1],
+                    "merkle": branch.iter().map(|h| h.to_string()).collect::<Vec<_>>(),
+                    "pos": pos,
+                })
             }
             other => panic!("unexpected request: {other}"),
-        }
+        })
     }
 }
 
@@ -73,27 +137,71 @@ fn drain_requests(
 ) -> Vec<Update<&'static str>> {
     let mut updates = Vec::new();
     while let Some(req) = queue.pop_front() {
-        let resp =
-            raw_msg(json!({ "jsonrpc": "2.0", "id": req.id, "result": server.answer(&req) }));
-        if let Some(update) = state.advance(queue, resp).expect("must advance") {
+        if let Some(update) = state
+            .advance(queue, response(&req, server))
+            .expect("must advance")
+        {
             updates.push(update);
         }
     }
     updates
 }
 
-/// A history response can report a confirmation height above the local tip: on a new block,
-/// romanz/electrs notifies the script hash before the header and answers requests in order, so
-/// the history arrives while the local tip is still one block behind. Such an anchor must be
-/// deferred until the tip catches up — not dropped — and must still be delivered without any
-/// further notification for that script.
-#[test]
-fn anchor_above_local_tip_is_deferred_until_tip_catches_up() -> anyhow::Result<()> {
+/// Drain like [`drain_requests`], but answer every merkle proof in the queue ahead of everything
+/// else in each round.
+///
+/// The Electrum protocol carries a request id precisely because responses need not come back in
+/// the order they were asked for: romanz/electrs happens to answer in order, Fulcrum processes
+/// requests concurrently. Nothing may depend on the ordering.
+fn drain_requests_proofs_first(
+    state: &mut BlockingState,
+    queue: &mut ReqQueue,
+    server: &Server,
+) -> Vec<Update<&'static str>> {
+    let mut updates = Vec::new();
+    while !queue.is_empty() {
+        let (proofs, rest): (Vec<_>, Vec<_>) = queue
+            .drain(..)
+            .partition(|req| req.method.as_ref() == "blockchain.transaction.get_merkle");
+        for req in proofs.into_iter().chain(rest) {
+            if let Some(update) = state
+                .advance(queue, response(&req, server))
+                .expect("must advance")
+            {
+                updates.push(update);
+            }
+        }
+    }
+    updates
+}
+
+/// The server's answer to `req`, as a raw JSON-RPC result or error message.
+fn response(req: &RawRequest, server: &Server) -> RawNotificationOrResponse {
+    raw_msg(match server.answer(req) {
+        Ok(result) => json!({ "jsonrpc": "2.0", "id": req.id, "result": result }),
+        Err(message) => json!({
+            "jsonrpc": "2.0",
+            "id": req.id,
+            "error": { "code": 1, "message": message },
+        }),
+    })
+}
+
+/// A descriptor to track, the script hash of its first spk, and a tx paying to that spk.
+fn tracked_descriptor() -> anyhow::Result<(
+    Descriptor<DescriptorPublicKey>,
+    ElectrumScriptHash,
+    ScriptBuf,
+)> {
     let descriptor = Descriptor::<DescriptorPublicKey>::from_str(&format!("wpkh({XPUB}/0/*)"))?;
     let spk = descriptor.at_derivation_index(0)?.script_pubkey();
     let spk_hash = ElectrumScriptHash::new(&spk);
+    Ok((descriptor, spk_hash, spk))
+}
 
-    let tx = Transaction {
+/// A coinbase paying `sats` to `spk`, so that varying `sats` gives a distinct tx.
+fn tx_paying(spk: &ScriptBuf, sats: u64) -> Transaction {
+    Transaction {
         version: transaction::Version::ONE,
         lock_time: absolute::LockTime::ZERO,
         input: vec![TxIn {
@@ -103,12 +211,14 @@ fn anchor_above_local_tip_is_deferred_until_tip_catches_up() -> anyhow::Result<(
             witness: Witness::new(),
         }],
         output: vec![TxOut {
-            value: Amount::from_sat(50_000),
-            script_pubkey: spk,
+            value: Amount::from_sat(sats),
+            script_pubkey: spk.clone(),
         }],
-    };
-    let txid = tx.compute_txid();
+    }
+}
 
+/// The regtest genesis block and an empty block on top of it.
+fn base_headers() -> (block::Header, block::Header) {
     let genesis = constants::genesis_block(Network::Regtest).header;
     let header_1 = block::Header {
         version: block::Version::ONE,
@@ -118,34 +228,90 @@ fn anchor_above_local_tip_is_deferred_until_tip_catches_up() -> anyhow::Result<(
         bits: CompactTarget::from_consensus(0x207fffff),
         nonce: 0,
     };
-    // With the tx at position 0 of a single-tx block, the merkle root is its txid.
-    let header_2 = block::Header {
-        merkle_root: Txid::to_raw_hash(txid).into(),
-        prev_blockhash: header_1.block_hash(),
-        time: 200,
-        ..header_1
-    };
+    (genesis, header_1)
+}
 
-    let mut cache = Cache::default();
-    cache.txs.insert(txid, Arc::new(tx.clone()));
+/// A block whose transactions have the given merkle root.
+fn block_with_root(
+    prev: &block::Header,
+    merkle_root: TxMerkleNode,
+    time: u32,
+    nonce: u32,
+) -> block::Header {
+    block::Header {
+        version: block::Version::ONE,
+        prev_blockhash: prev.block_hash(),
+        merkle_root,
+        time,
+        bits: CompactTarget::from_consensus(0x207fffff),
+        nonce,
+    }
+}
 
-    let mut spk_tracker = DerivedSpkTracker::new(0);
-    spk_tracker.insert_descriptor("external", descriptor, 0);
+/// A block whose only transaction is `txid`, so that its merkle root is the txid itself.
+fn block_with_tx(prev: &block::Header, txid: Txid, time: u32, nonce: u32) -> block::Header {
+    block_with_root(prev, Txid::to_raw_hash(txid).into(), time, nonce)
+}
 
-    let mut state = BlockingState::new(
-        ReqCoord::default(),
+fn new_state(
+    cache: Cache,
+    descriptor: Descriptor<DescriptorPublicKey>,
+    genesis: block::Header,
+) -> BlockingState {
+    new_state_with_cp(
         cache,
-        spk_tracker,
+        descriptor,
         CheckPoint::new(BlockId {
             height: 0,
             hash: genesis.block_hash(),
         }),
-    );
+    )
+}
+
+fn new_state_with_cp(
+    cache: Cache,
+    descriptor: Descriptor<DescriptorPublicKey>,
+    cp: CheckPoint,
+) -> BlockingState {
+    let mut spk_tracker = DerivedSpkTracker::new(0);
+    spk_tracker.insert_descriptor("external", descriptor, 0);
+    BlockingState::new(ReqCoord::default(), cache, spk_tracker, cp)
+}
+
+/// The anchor a tx confirmed in `header` at `height` must be given.
+fn anchor_of(header: &block::Header, height: u32) -> ConfirmationBlockTime {
+    ConfirmationBlockTime {
+        block_id: BlockId {
+            height,
+            hash: header.block_hash(),
+        },
+        confirmation_time: header.time as u64,
+    }
+}
+
+/// A history response can report a confirmation height above the local tip: on a new block,
+/// romanz/electrs notifies the script hash before the header and answers requests in order, so
+/// the history arrives while the local tip is still one block behind. Such an anchor must be
+/// deferred until the tip catches up — not dropped — and must still be delivered without any
+/// further notification for that script.
+#[test]
+fn anchor_above_local_tip_is_deferred_until_tip_catches_up() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+    let header_2 = block_with_tx(&header_1, txid, 200, 0);
+
+    let mut cache = Cache::default();
+    cache.txs.insert(txid, Arc::new(tx.clone()));
+
+    let mut state = new_state(cache, descriptor, genesis);
     let mut queue = ReqQueue::new();
     let mut server = Server {
         headers: vec![genesis, header_1],
         spk_hash,
-        tx,
+        txs: vec![(tx, 2)],
+        merkle_proof: (Vec::new(), 0),
     };
 
     state.init(&mut queue);
@@ -185,17 +351,11 @@ fn anchor_above_local_tip_is_deferred_until_tip_catches_up() -> anyhow::Result<(
     server.headers.push(header_2);
     let updates = drain_requests(&mut state, &mut queue, &server);
 
-    let expected_anchor = ConfirmationBlockTime {
-        block_id: BlockId {
-            height: 2,
-            hash: header_2.block_hash(),
-        },
-        confirmation_time: header_2.time as u64,
-    };
     assert!(
-        updates
-            .iter()
-            .any(|u| u.tx_update.anchors.contains(&(expected_anchor, txid))),
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2, 2), txid))),
         "anchor must be delivered once the tip catches up"
     );
     Ok(())
@@ -234,3 +394,1225 @@ fn descriptor_inserted_mid_connection_is_subscribed() -> anyhow::Result<()> {
     );
     Ok(())
 }
+
+/// A reorg can move a transaction into a different block at the *same* height. An Electrum
+/// script status is a hash over txid-height pairs, so it does not change and the server has no
+/// reason to send a script hash notification. The anchor we already delivered now points at a
+/// block that is no longer in the chain, so it must be refetched off the tip update alone.
+#[test]
+fn anchor_is_refetched_when_tx_moves_to_another_block_of_same_height() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+    let header_2 = block_with_tx(&header_1, txid, 200, 0);
+    // The block that replaces height 2 contains the tx too, hence the identical script status.
+    let header_2b = block_with_tx(&header_1, txid, 222, 1);
+    let header_3b = block::Header {
+        prev_blockhash: header_2b.block_hash(),
+        time: 300,
+        ..header_1
+    };
+    assert_ne!(header_2.block_hash(), header_2b.block_hash());
+
+    let mut state = new_state(Cache::default(), descriptor, genesis);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: vec![genesis, header_1, header_2],
+        spk_hash,
+        txs: vec![(tx, 2)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    state.init(&mut queue);
+    let updates = drain_requests(&mut state, &mut queue, &server);
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2, 2), txid))),
+        "tx must first be anchored to the original block"
+    );
+
+    // Reorg. Only a header notification is sent: the script status is unchanged, so a server
+    // has no reason to notify the script hash.
+    server.headers = vec![genesis, header_1, header_2b, header_3b];
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.headers.subscribe",
+            "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }],
+        })),
+    )?;
+    let updates = drain_requests(&mut state, &mut queue, &server);
+
+    assert!(
+        updates.iter().any(|u| u
+            .chain_update
+            .as_ref()
+            .is_some_and(|cp| cp.block_id() == anchor_of(&header_3b, 3).block_id)),
+        "chain update must follow the reorg"
+    );
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2b, 2), txid))),
+        "anchor must be refetched for the block that replaced the evicted one"
+    );
+    Ok(())
+}
+
+/// A reorg can land while an anchor fetch is in flight. The merkle proof we get back was built
+/// against the chain the server had when it received the request, so it may not prove inclusion
+/// in the block we now have at that height. Verifying it against that block would record a
+/// permanent "not in this block" verdict for an anchor which is in fact valid.
+#[test]
+fn merkle_proof_predating_a_reorg_is_not_taken_as_a_failed_anchor() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+    let header_2 = block_with_tx(&header_1, txid, 200, 0);
+    // The block which replaces height 2 contains the tx alongside another one, so the tx keeps
+    // its height — and with it its script status — but needs a different merkle proof.
+    let proof_2b = response::TxMerkle {
+        block_height: absolute::Height::from_consensus(2)?,
+        merkle: vec![sha256d::Hash::hash(b"the other tx")],
+        pos: 1,
+    };
+    let header_2b = block_with_root(&header_1, proof_2b.expected_merkle_root(txid), 222, 1);
+    let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 300, 0);
+
+    let mut state = new_state(Cache::default(), descriptor, genesis);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: vec![genesis, header_1, header_2],
+        spk_hash,
+        txs: vec![(tx, 2)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    // Sync, but hold back the merkle proof so the anchor fetch is still in flight.
+    state.init(&mut queue);
+    let mut in_flight = Vec::new();
+    while let Some(req) = queue.pop_front() {
+        if req.method.as_ref() == "blockchain.transaction.get_merkle" {
+            in_flight.push(req);
+            continue;
+        }
+        state.advance(&mut queue, response(&req, &server))?;
+    }
+    let stale_req = match in_flight.as_slice() {
+        [req] => req.clone(),
+        reqs => panic!(
+            "expected exactly one merkle request in flight, got {}",
+            reqs.len()
+        ),
+    };
+    let stale_resp = response(&stale_req, &server);
+
+    // The reorg lands before the server answers.
+    server.headers = vec![genesis, header_1, header_2b, header_3b];
+    server.merkle_proof = (proof_2b.merkle.clone(), proof_2b.pos);
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.headers.subscribe",
+            "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }],
+        })),
+    )?;
+    drain_requests(&mut state, &mut queue, &server);
+
+    // The held answer proves inclusion in the block which was evicted, not in the one which
+    // replaced it.
+    state.advance(&mut queue, stale_resp)?;
+    let updates = drain_requests(&mut state, &mut queue, &server);
+
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2b, 2), txid))),
+        "the anchor must be refetched rather than written off from a proof of the evicted block"
+    );
+    Ok(())
+}
+
+/// A reorg can land while a job is midway through fetching anchors. Anchors are staged as they
+/// resolve, so the job must give up the ones it staged against the chain it started on — otherwise
+/// it goes on to emit them in a single update alongside the ones it resolved against the chain it
+/// ended on.
+#[test]
+fn anchors_staged_before_a_reorg_are_not_emitted_after_it() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let (tx_a, tx_b) = (tx_paying(&spk, 50_000), tx_paying(&spk, 60_000));
+    let (txid_a, txid_b) = (tx_a.compute_txid(), tx_b.compute_txid());
+    let (genesis, header_1) = base_headers();
+    let header_2 = block_with_tx(&header_1, txid_a, 200, 0);
+    let header_3 = block_with_tx(&header_2, txid_b, 300, 0);
+    // The reorg keeps both txs at their heights — hence the unchanged script status — but in
+    // different blocks, and extends the chain by one.
+    let header_2b = block_with_tx(&header_1, txid_a, 222, 1);
+    let header_3b = block_with_tx(&header_2b, txid_b, 333, 1);
+    let header_4b = block_with_root(&header_3b, TxMerkleNode::all_zeros(), 400, 0);
+
+    let mut state = new_state(Cache::default(), descriptor, genesis);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: vec![genesis, header_1, header_2, header_3],
+        spk_hash,
+        txs: vec![(tx_a, 2), (tx_b, 3)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    // Sync, but hold back tx_b's proof so that the job has staged tx_a's anchor and is still
+    // waiting on tx_b's when the reorg lands.
+    state.init(&mut queue);
+    let mut in_flight = Vec::new();
+    while let Some(req) = queue.pop_front() {
+        if req.method.as_ref() == "blockchain.transaction.get_merkle"
+            && req.params[0] == json!(txid_b.to_string())
+        {
+            in_flight.push(req);
+            continue;
+        }
+        state.advance(&mut queue, response(&req, &server))?;
+    }
+    let held_req = match in_flight.as_slice() {
+        [req] => req.clone(),
+        reqs => panic!("expected one held merkle request, got {}", reqs.len()),
+    };
+    let held_resp = response(&held_req, &server);
+
+    server.headers = vec![genesis, header_1, header_2b, header_3b, header_4b];
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.headers.subscribe",
+            "params": [{ "hex": serialize_hex(&header_4b), "height": 4 }],
+        })),
+    )?;
+    let mut updates = drain_requests(&mut state, &mut queue, &server);
+    updates.extend(state.advance(&mut queue, held_resp)?);
+    updates.extend(drain_requests(&mut state, &mut queue, &server));
+
+    let anchors = updates
+        .iter()
+        .flat_map(|u| u.tx_update.anchors.iter().copied())
+        .collect::<Vec<_>>();
+    for evicted in [
+        (anchor_of(&header_2, 2), txid_a),
+        (anchor_of(&header_3, 3), txid_b),
+    ] {
+        assert!(
+            !anchors.contains(&evicted),
+            "an anchor to an evicted block must not be emitted after the reorg: {evicted:?}"
+        );
+    }
+    for expected in [
+        (anchor_of(&header_2b, 2), txid_a),
+        (anchor_of(&header_3b, 3), txid_b),
+    ] {
+        assert!(
+            anchors.contains(&expected),
+            "both anchors must be refetched against the new chain: {expected:?}"
+        );
+    }
+    Ok(())
+}
+
+/// The everyday reorg: one which takes a transaction out of its block and back to the mempool.
+/// The refetch is speculative — we ask for a proof of inclusion at a height the transaction was
+/// *last seen* at — so the server answering "not in that block" is expected, and must not take
+/// the connection down with it.
+#[test]
+fn a_tx_unconfirmed_by_a_reorg_does_not_error_the_connection() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+    let header_2 = block_with_tx(&header_1, txid, 200, 0);
+    // Height 2 is replaced by a block without the tx, and the chain grows by one.
+    let header_2b = block_with_root(&header_1, TxMerkleNode::all_zeros(), 222, 1);
+    let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 333, 0);
+
+    let mut state = new_state(Cache::default(), descriptor, genesis);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: vec![genesis, header_1, header_2],
+        spk_hash,
+        txs: vec![(tx, 2)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    state.init(&mut queue);
+    let updates = drain_requests(&mut state, &mut queue, &server);
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2, 2), txid))),
+        "tx must first be anchored"
+    );
+
+    // The reorg leaves the tx in the mempool, so the server no longer has it in any block.
+    server.headers = vec![genesis, header_1, header_2b, header_3b];
+    server.txs = Vec::new();
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.headers.subscribe",
+            "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }],
+        })),
+    )?;
+
+    while let Some(req) = queue.pop_front() {
+        state
+            .advance(&mut queue, response(&req, &server))
+            .map_err(|e| anyhow::anyhow!("{e:#}"))?;
+    }
+    Ok(())
+}
+
+/// The other half of `merkle_proof_predating_a_reorg_is_not_taken_as_a_failed_anchor`: a server
+/// answers a proof request from the chain it had when it *received* it, so if the tx was out of
+/// its block at that moment it answers with an error — an error about the chain we have since
+/// left. Blaming that on whichever block the reorg put at the height would write off an anchor
+/// which is in fact valid.
+#[test]
+fn merkle_error_predating_a_reorg_is_not_taken_as_a_failed_anchor() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+    let header_2 = block_with_tx(&header_1, txid, 200, 0);
+    // Height 2 is replaced by a block which contains the tx too, so the anchor is still valid.
+    let header_2b = block_with_tx(&header_1, txid, 222, 1);
+    let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 300, 0);
+
+    let mut state = new_state(Cache::default(), descriptor, genesis);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: vec![genesis, header_1, header_2],
+        spk_hash,
+        txs: vec![(tx, 2)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    // Sync, but hold back the merkle request so the anchor fetch is still in flight.
+    state.init(&mut queue);
+    let mut in_flight = Vec::new();
+    while let Some(req) = queue.pop_front() {
+        if req.method.as_ref() == "blockchain.transaction.get_merkle" {
+            in_flight.push(req);
+            continue;
+        }
+        state.advance(&mut queue, response(&req, &server))?;
+    }
+    let stale_req = match in_flight.as_slice() {
+        [req] => req.clone(),
+        reqs => panic!(
+            "expected exactly one merkle request in flight, got {}",
+            reqs.len()
+        ),
+    };
+
+    // The reorg lands before the server answers.
+    server.headers = vec![genesis, header_1, header_2b, header_3b];
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.headers.subscribe",
+            "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }],
+        })),
+    )?;
+    drain_requests(&mut state, &mut queue, &server);
+
+    // The held request was received while the tx was out of its block, so it is answered with
+    // an error — the wording is the one romanz/electrs really sends, a bare JSON string which
+    // conflates a genuine fault with the everyday reorg.
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "id": stale_req.id,
+            "error": "tx not found or is unconfirmed",
+        })),
+    )?;
+    let updates = drain_requests(&mut state, &mut queue, &server);
+
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2b, 2), txid))),
+        "the anchor must be refetched rather than written off from an error about the evicted block"
+    );
+    Ok(())
+}
+
+/// A server error is not a disproof — it is equally a rate limit, an index still catching up or
+/// a daemon hiccup — so it must not be recorded as one. The transaction stays in the record of
+/// what was seen at that height, so the next reorg of that height asks again; and the job it
+/// blocked must still finish rather than re-ask in a loop.
+#[test]
+fn a_merkle_error_is_not_recorded_as_a_failed_anchor() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+    let header_2 = block_with_tx(&header_1, txid, 200, 0);
+    // Height 2 is replaced by a block containing the tx, and the chain grows by one.
+    let header_2b = block_with_tx(&header_1, txid, 222, 1);
+    let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 300, 0);
+
+    let mut state = new_state(Cache::default(), descriptor, genesis);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: vec![genesis, header_1, header_2],
+        spk_hash,
+        txs: vec![(tx.clone(), 2)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    // Sync, but fail every merkle request the way a busy server would.
+    state.init(&mut queue);
+    let mut merkle_requests = 0;
+    while let Some(req) = queue.pop_front() {
+        let resp = if req.method.as_ref() == "blockchain.transaction.get_merkle" {
+            merkle_requests += 1;
+            assert!(
+                merkle_requests < 10,
+                "a server error must not put the job in a re-ask loop"
+            );
+            raw_msg(json!({
+                "jsonrpc": "2.0",
+                "id": req.id,
+                "error": { "code": 1, "message": "server busy" },
+            }))
+        } else {
+            response(&req, &server)
+        };
+        state.advance(&mut queue, resp)?;
+    }
+    assert_eq!(
+        merkle_requests, 1,
+        "the job must give up on the pair rather than re-ask"
+    );
+    assert!(
+        state.cache().anchors.is_empty(),
+        "an error proves nothing, so no anchor may be recorded from it"
+    );
+
+    // The reorg replaces the block, so the anchor is asked for again — which it could not be
+    // had the error dropped this script from `spk_hashes_by_height`.
+    server.headers = vec![genesis, header_1, header_2b, header_3b];
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.headers.subscribe",
+            "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }],
+        })),
+    )?;
+    let updates = drain_requests(&mut state, &mut queue, &server);
+
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2b, 2), txid))),
+        "the tx must still be asked about at this height after a server error"
+    );
+    Ok(())
+}
+
+/// Issue #12's literal case: a reorg to a block of the *same* height, with no growth at all.
+/// `ChainJob` applies this by short-circuit straight from the header notification, so it is the
+/// one reorg shape which never fetches anything — and every other reorg test here also grows the
+/// chain, which takes a different path.
+#[test]
+fn anchor_is_refetched_after_a_same_height_reorg() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+    let header_2 = block_with_tx(&header_1, txid, 200, 0);
+    let header_2b = block_with_tx(&header_1, txid, 222, 1);
+    assert_ne!(header_2.block_hash(), header_2b.block_hash());
+
+    let mut state = new_state(Cache::default(), descriptor, genesis);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: vec![genesis, header_1, header_2],
+        spk_hash,
+        txs: vec![(tx, 2)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    state.init(&mut queue);
+    let updates = drain_requests(&mut state, &mut queue, &server);
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2, 2), txid))),
+        "tx must first be anchored to the original block"
+    );
+
+    // The tip does not move: same height, different block, unchanged script status.
+    server.headers = vec![genesis, header_1, header_2b];
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.headers.subscribe",
+            "params": [{ "hex": serialize_hex(&header_2b), "height": 2 }],
+        })),
+    )?;
+    let updates = drain_requests(&mut state, &mut queue, &server);
+
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2b, 2), txid))),
+        "anchor must be refetched for the block that replaced the evicted one"
+    );
+    Ok(())
+}
+
+/// A proof is verified against the merkle root of the block we have at that height, so a job
+/// which asks for the proof and that block's header together only works if the server answers in
+/// request order. It need not: the protocol carries request ids for that reason.
+#[test]
+fn anchor_is_refetched_whatever_order_the_server_answers_in() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+    let header_2 = block_with_tx(&header_1, txid, 200, 0);
+    // Height 2 is replaced and the chain grows, so the notified header is height 3's — the
+    // replacement block's header has to be fetched before its proof can be verified.
+    let header_2b = block_with_tx(&header_1, txid, 222, 1);
+    let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 300, 0);
+
+    let mut state = new_state(Cache::default(), descriptor, genesis);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: vec![genesis, header_1, header_2],
+        spk_hash,
+        txs: vec![(tx, 2)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    state.init(&mut queue);
+    let updates = drain_requests(&mut state, &mut queue, &server);
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2, 2), txid))),
+        "tx must first be anchored to the original block"
+    );
+
+    server.headers = vec![genesis, header_1, header_2b, header_3b];
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.headers.subscribe",
+            "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }],
+        })),
+    )?;
+    let updates = drain_requests_proofs_first(&mut state, &mut queue, &server);
+
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2b, 2), txid))),
+        "the anchor must be refetched even when the proof overtakes the header"
+    );
+    Ok(())
+}
+
+/// A client restored from a persisted checkpoint chain starts with blocks in its chain whose
+/// headers are not in its cache. Resolving an anchor at such a height needs both the header and
+/// the proof, and nothing else will fetch that header — `ChainJob` short-circuits, since the tip
+/// is already correct.
+///
+/// So this is the case where the two halves of the ordering fix are load-bearing: the proof must
+/// not be asked for before the header is cached, and the header response must advance the waiting
+/// job even though the chain itself has nothing to learn from it.
+#[test]
+fn anchor_resolves_when_the_chain_is_restored_without_its_headers() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+    let header_2 = block_with_tx(&header_1, txid, 200, 0);
+    let header_3 = block_with_root(&header_2, TxMerkleNode::all_zeros(), 300, 0);
+
+    // The restored chain knows the blocks, the fresh cache knows none of their headers. The tx
+    // is one block below the tip, so the header it needs is not the one `headers.subscribe`
+    // hands back and nothing else fetches it either — `ChainJob` short-circuits, the tip being
+    // already correct.
+    let cp = CheckPoint::new(BlockId {
+        height: 0,
+        hash: genesis.block_hash(),
+    })
+    .insert(BlockId {
+        height: 2,
+        hash: header_2.block_hash(),
+    })
+    .insert(BlockId {
+        height: 3,
+        hash: header_3.block_hash(),
+    });
+    let mut state = new_state_with_cp(Cache::default(), descriptor, cp);
+    let mut queue = ReqQueue::new();
+    let server = Server {
+        headers: vec![genesis, header_1, header_2, header_3],
+        spk_hash,
+        txs: vec![(tx, 2)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    state.init(&mut queue);
+    let updates = drain_requests_proofs_first(&mut state, &mut queue, &server);
+
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2, 2), txid))),
+        "the anchor must resolve even when the proof overtakes the header it is verified against"
+    );
+    Ok(())
+}
+
+/// A header fetched before a reorg describes the chain we have since left behind, and inserting
+/// it would splice a purged block into the checkpoint chain.
+///
+/// `extends`/`replaces` do not catch this on their own: they only decline a height the chain
+/// already has. The gap they leave open is a *sparse* chain — a restored one, or one whose
+/// missing heights sit below the 21-block suffix `ChainJob` rewrites — reorged deeper than that
+/// suffix, so the refetch never learns the low block changed too.
+#[test]
+fn header_fetched_before_a_reorg_is_not_spliced_into_the_chain() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+
+    // Two chains which differ at height 2 as well as near the tip. Both contain the tx at
+    // height 2, so the anchor stays valid throughout — only the block it belongs to changes.
+    let build = |second: block::Header, tip: u32, nonce: u32| {
+        let mut chain = vec![genesis, header_1, second];
+        for height in 3..=tip {
+            let prev = *chain.last().expect("non-empty");
+            chain.push(block_with_root(
+                &prev,
+                TxMerkleNode::all_zeros(),
+                1000 + height,
+                nonce,
+            ));
+        }
+        chain
+    };
+    let chain_a = build(block_with_tx(&header_1, txid, 200, 0), 30, 0);
+    let chain_b = build(block_with_tx(&header_1, txid, 222, 1), 31, 1);
+    let (a2, b2) = (chain_a[2], chain_b[2]);
+    assert_ne!(a2.block_hash(), b2.block_hash());
+
+    // A restored chain sparse enough that height 2 is a gap — so the anchor has to fetch that
+    // header, and `replaces` will not decline it when it comes back.
+    let cp = CheckPoint::new(BlockId {
+        height: 0,
+        hash: genesis.block_hash(),
+    })
+    .insert(BlockId {
+        height: 30,
+        hash: chain_a[30].block_hash(),
+    });
+    let mut state = new_state_with_cp(Cache::default(), descriptor, cp);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: chain_a.clone(),
+        spk_hash,
+        txs: vec![(tx, 2)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    // Sync, but hold back the height-2 header so the fetch is still in flight.
+    state.init(&mut queue);
+    let mut in_flight = Vec::new();
+    while let Some(req) = queue.pop_front() {
+        if req.method.as_ref() == "blockchain.block.header" && req.params[0] == json!(2) {
+            in_flight.push(req);
+            continue;
+        }
+        state.advance(&mut queue, response(&req, &server))?;
+    }
+    let stale_req = match in_flight.as_slice() {
+        [req] => req.clone(),
+        reqs => panic!("expected one held header request, got {}", reqs.len()),
+    };
+    let stale_resp = response(&stale_req, &server);
+
+    // The reorg lands. It runs deeper than `ChainJob`'s suffix, so the refetch rewrites the
+    // top 21 blocks and never learns that height 2 changed too.
+    server.headers = chain_b.clone();
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.headers.subscribe",
+            "params": [{ "hex": serialize_hex(&chain_b[31]), "height": 31 }],
+        })),
+    )?;
+    drain_requests(&mut state, &mut queue, &server);
+
+    // The held answer describes the chain we have left behind.
+    let mut updates = Vec::from_iter(state.advance(&mut queue, stale_resp)?);
+    updates.extend(drain_requests(&mut state, &mut queue, &server));
+
+    let tip = updates
+        .iter()
+        .rev()
+        .find_map(|u| u.chain_update.clone())
+        .expect("must get a chain update");
+    let at_2 = tip.iter().find(|cp| cp.height() == 2);
+    assert_ne!(
+        at_2.as_ref().map(|cp| cp.hash()),
+        Some(a2.block_hash()),
+        "a header from the chain we left must not be spliced into the checkpoint chain"
+    );
+    assert_eq!(
+        at_2.map(|cp| cp.hash()),
+        Some(b2.block_hash()),
+        "the height must be refetched against the chain we are actually on"
+    );
+    assert!(
+        updates
+            .iter()
+            .any(|u| u.tx_update.anchors.contains(&(anchor_of(&b2, 2), txid))),
+        "and the anchor must resolve against that block"
+    );
+    Ok(())
+}
+
+/// The refetch is a script hash notification we raise ourselves, so it must not displace one the
+/// server actually sent. A real notification carries a status at least as new as anything we
+/// could replay from cache; replacing its job would resolve the script against a stale history
+/// and drop whatever the new status was reporting.
+#[test]
+fn a_replayed_job_does_not_displace_one_the_server_started() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let (tx_a, tx_b) = (tx_paying(&spk, 50_000), tx_paying(&spk, 60_000));
+    let (txid_a, txid_b) = (tx_a.compute_txid(), tx_b.compute_txid());
+    let (genesis, _) = base_headers();
+    let header_1 = block_with_tx(&genesis, txid_b, 100, 0);
+    let header_2 = block_with_tx(&header_1, txid_a, 200, 0);
+    // The reorg replaces height 2 with another block holding tx_a, and extends the chain.
+    let header_2b = block_with_tx(&header_1, txid_a, 222, 1);
+    let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 300, 0);
+
+    let mut state = new_state(Cache::default(), descriptor, genesis);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: vec![genesis, header_1, header_2],
+        spk_hash,
+        // The server only reports tx_a to begin with.
+        txs: vec![(tx_a, 2)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    state.init(&mut queue);
+    let updates = drain_requests(&mut state, &mut queue, &server);
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2, 2), txid_a))),
+        "tx_a must first be anchored to the original block"
+    );
+
+    // The server now reports tx_b as well, and notifies the new status. The job that starts is
+    // the only thing which knows about tx_b.
+    server.txs.insert(0, (tx_b, 1));
+    let new_status =
+        ElectrumScriptStatus::from_history(&server.history(&json!(spk_hash.to_string())))
+            .expect("history must be non-empty");
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.scripthash.subscribe",
+            "params": [spk_hash.to_string(), new_status.to_string()],
+        })),
+    )?;
+
+    // Hold back that job's history, so it is still in flight when the reorg lands.
+    let mut in_flight = Vec::new();
+    while let Some(req) = queue.pop_front() {
+        if req.method.as_ref() == "blockchain.scripthash.get_history" {
+            in_flight.push(req);
+            continue;
+        }
+        state.advance(&mut queue, response(&req, &server))?;
+    }
+    let held_req = match in_flight.as_slice() {
+        [req] => req.clone(),
+        reqs => panic!("expected one held history request, got {}", reqs.len()),
+    };
+
+    // The reorg evicts height 2, where this script is recorded — so the refetch wants to replay
+    // its job, and must decline because the server's own job is already there.
+    server.headers = vec![genesis, header_1, header_2b, header_3b];
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.headers.subscribe",
+            "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }],
+        })),
+    )?;
+    let mut updates = drain_requests(&mut state, &mut queue, &server);
+    updates.extend(state.advance(&mut queue, response(&held_req, &server))?);
+    updates.extend(drain_requests(&mut state, &mut queue, &server));
+
+    assert!(
+        updates
+            .iter()
+            .any(|u| u.tx_update.txs.iter().any(|tx| tx.compute_txid() == txid_b)),
+        "the server's own job must survive and deliver what its status was reporting"
+    );
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2b, 2), txid_a))),
+        "and must still refetch the anchor the reorg invalidated"
+    );
+    Ok(())
+}
+
+/// A persisted checkpoint chain can be stale at a height below the window [`ChainJob`] rewrites
+/// — an offline reorg, say. Then the block *we* have at that height is one the server does not
+/// have, and `blockchain.block.header` is keyed by height, so no request can ever fetch it.
+///
+/// Withholding the proof until that header is cached must not turn into an endless request loop.
+#[test]
+fn a_header_the_server_does_not_have_does_not_loop() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+    // The server's height 2, and the stale one our persisted chain still claims.
+    let server_2 = block_with_tx(&header_1, txid, 200, 0);
+    let stale_2 = block_with_tx(&header_1, txid, 999, 7);
+    assert_ne!(server_2.block_hash(), stale_2.block_hash());
+
+    let mut chain = vec![genesis, header_1, server_2];
+    for height in 3..=30u32 {
+        let prev = *chain.last().expect("non-empty");
+        chain.push(block_with_root(
+            &prev,
+            TxMerkleNode::all_zeros(),
+            1000 + height,
+            0,
+        ));
+    }
+
+    // Tip agrees with the server, so `ChainJob` short-circuits and never rewrites height 2.
+    let cp = CheckPoint::new(BlockId {
+        height: 0,
+        hash: genesis.block_hash(),
+    })
+    .insert(BlockId {
+        height: 2,
+        hash: stale_2.block_hash(),
+    })
+    .insert(BlockId {
+        height: 30,
+        hash: chain[30].block_hash(),
+    });
+    let mut state = new_state_with_cp(Cache::default(), descriptor, cp);
+    let mut queue = ReqQueue::new();
+    let server = Server {
+        headers: chain,
+        spk_hash,
+        txs: vec![(tx, 2)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    state.init(&mut queue);
+    let mut served = 0;
+    while let Some(req) = queue.pop_front() {
+        served += 1;
+        assert!(
+            served < 200,
+            "the client must not loop: {served} requests, last was {} {:?}",
+            req.method,
+            req.params
+        );
+        state.advance(&mut queue, response(&req, &server))?;
+    }
+    Ok(())
+}
+
+/// A replay is built from the last status and the heights it was seen at, so both have to go
+/// when the server stops reporting a history for the script — an RBF'd transaction, say. Left
+/// behind, a later eviction at that height would rebuild the job from a history the script no
+/// longer has and go asking for proofs of a transaction that is gone.
+#[test]
+fn a_script_whose_history_goes_away_is_not_replayed() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+    let header_2 = block_with_tx(&header_1, txid, 200, 0);
+    let header_2b = block_with_root(&header_1, TxMerkleNode::all_zeros(), 222, 1);
+    let header_3b = block_with_root(&header_2b, TxMerkleNode::all_zeros(), 300, 0);
+
+    let mut state = new_state(Cache::default(), descriptor, genesis);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: vec![genesis, header_1, header_2],
+        spk_hash,
+        txs: vec![(tx, 2)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    state.init(&mut queue);
+    let updates = drain_requests(&mut state, &mut queue, &server);
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2, 2), txid))),
+        "tx must first be anchored"
+    );
+    assert!(
+        state.spk_histories().status(spk_hash).is_some(),
+        "the status must be recorded while the script has a history"
+    );
+
+    // The transaction is gone, so the script's status goes to null.
+    server.txs = Vec::new();
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.scripthash.subscribe",
+            "params": [spk_hash.to_string(), null],
+        })),
+    )?;
+    drain_requests(&mut state, &mut queue, &server);
+    assert!(
+        !state.spk_histories().status(spk_hash).is_some(),
+        "a null status must drop the recorded status"
+    );
+
+    // A reorg evicting the height it used to be seen at must not replay anything.
+    server.headers = vec![genesis, header_1, header_2b, header_3b];
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.headers.subscribe",
+            "params": [{ "hex": serialize_hex(&header_3b), "height": 3 }],
+        })),
+    )?;
+    let mut asked = Vec::new();
+    while let Some(req) = queue.pop_front() {
+        asked.push(req.method.to_string());
+        state.advance(&mut queue, response(&req, &server))?;
+    }
+    assert!(
+        !asked
+            .iter()
+            .any(|m| m == "blockchain.transaction.get_merkle"),
+        "no proof may be asked for a transaction the script no longer has, got: {asked:?}"
+    );
+    Ok(())
+}
+
+/// A server proves inclusion in whichever block *it* has at a height, so a proof whose root does
+/// not match ours says the two chains disagree there — not that the transaction is absent from
+/// our block. Remembering that against our block would be a verdict the proof cannot support, and
+/// a chain that came back to that block would consult it and skip the anchor for good.
+#[test]
+fn a_proof_for_another_block_is_not_a_verdict_on_ours() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+    let ours = block_with_tx(&header_1, txid, 200, 0);
+    // Their block holds the tx alongside another, so it needs a different proof — the root the
+    // server's proof expands to cannot match the root of our block.
+    let proof_theirs = response::TxMerkle {
+        block_height: absolute::Height::from_consensus(2)?,
+        merkle: vec![sha256d::Hash::hash(b"the other tx")],
+        pos: 1,
+    };
+    let theirs = block_with_root(&header_1, proof_theirs.expected_merkle_root(txid), 222, 1);
+    assert_ne!(ours.merkle_root, theirs.merkle_root);
+
+    let mut chain = vec![genesis, header_1, theirs];
+    for height in 3..=30u32 {
+        let prev = *chain.last().expect("non-empty");
+        chain.push(block_with_root(
+            &prev,
+            TxMerkleNode::all_zeros(),
+            1000 + height,
+            0,
+        ));
+    }
+
+    // A persisted chain holding our block at height 2, and its header already cached — otherwise
+    // `GetHeader` catches the disagreement before any proof is asked for. The tip agrees, so no
+    // chain job runs and nothing rewrites height 2: the disagreement is below the window.
+    let mut cache = Cache::default();
+    cache.headers.insert(ours.block_hash(), ours);
+    let cp = CheckPoint::new(BlockId {
+        height: 0,
+        hash: genesis.block_hash(),
+    })
+    .insert(BlockId {
+        height: 2,
+        hash: ours.block_hash(),
+    })
+    .insert(BlockId {
+        height: 30,
+        hash: chain[30].block_hash(),
+    });
+    let mut state = new_state_with_cp(cache, descriptor, cp);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: chain,
+        spk_hash,
+        txs: vec![(tx, 2)],
+        merkle_proof: (proof_theirs.merkle.clone(), proof_theirs.pos),
+    };
+
+    state.init(&mut queue);
+    let mut served = 0;
+    while let Some(req) = queue.pop_front() {
+        served += 1;
+        assert!(served < 200, "a mismatch must not become a request loop");
+        state.advance(&mut queue, response(&req, &server))?;
+    }
+    assert!(
+        state.cache().anchors.is_empty(),
+        "a proof for a block we do not have must not anchor anything"
+    );
+
+    // The server comes back to our block at that height. Nothing durable was written against it,
+    // so the job a notification rebuilds must be able to anchor there.
+    server.headers[2] = ours;
+    server.merkle_proof = (Vec::new(), 0);
+    let status = ElectrumScriptStatus::from_history(&server.history(&json!(spk_hash.to_string())))
+        .expect("history must be non-empty");
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.scripthash.subscribe",
+            "params": [spk_hash.to_string(), status.to_string()],
+        })),
+    )?;
+    let updates = drain_requests(&mut state, &mut queue, &server);
+    assert!(
+        updates
+            .iter()
+            .any(|u| u.tx_update.anchors.contains(&(anchor_of(&ours, 2), txid))),
+        "the anchor must still be reachable once the chains agree again"
+    );
+    Ok(())
+}
+
+/// A job needs both the transactions and their anchors, and the server answers in any order.
+///
+/// Every other test lets the `GetTx` land first, which puts the anchors on the job's final pass
+/// with nothing running after them. This forces the other order: the merkle proof arrives first,
+/// so the anchor resolves early and the job runs again when the transaction finally lands.
+///
+/// Anchors are re-resolved from scratch on every pass, so that later pass must not lose the
+/// anchor already in hand — the set of what to anchor is the question, not the answer, and
+/// clearing it once answered would have the next pass ask an empty question and stage nothing.
+#[test]
+fn anchor_survives_a_pass_that_happens_after_it_resolved() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, spk) = tracked_descriptor()?;
+    let tx = tx_paying(&spk, 50_000);
+    let txid = tx.compute_txid();
+    let (genesis, header_1) = base_headers();
+    let header_2 = block_with_tx(&header_1, txid, 200, 0);
+
+    let mut state = new_state(Cache::default(), descriptor, genesis);
+    let mut queue = ReqQueue::new();
+    let server = Server {
+        headers: vec![genesis, header_1, header_2],
+        spk_hash,
+        txs: vec![(tx, 2)],
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    state.init(&mut queue);
+    let updates = drain_requests_proofs_first(&mut state, &mut queue, &server);
+    assert!(
+        updates.iter().any(|u| u
+            .tx_update
+            .anchors
+            .contains(&(anchor_of(&header_2, 2), txid))),
+        "the anchor must still be emitted after a later pass"
+    );
+    Ok(())
+}
+
+/// A tip notification landing while the previous one's headers are still in flight.
+///
+/// The replacement job asks for the same heights, so its request is byte-identical to the one
+/// already out. Deduplicated against that one, nothing new is sent, and the answer already on its
+/// way describes the chain the server has just left.
+#[test]
+fn a_tip_that_moves_while_headers_are_in_flight_is_not_lost() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, _spk) = tracked_descriptor()?;
+    let (genesis, header_1) = base_headers();
+    let build = |nonce: u32| {
+        let mut chain = vec![genesis, header_1];
+        for height in 2..=3 {
+            let prev = *chain.last().expect("non-empty");
+            chain.push(block_with_root(
+                &prev,
+                TxMerkleNode::all_zeros(),
+                1000 + height,
+                nonce,
+            ));
+        }
+        chain
+    };
+    let (chain_a, chain_b) = (build(0), build(1));
+    assert_ne!(chain_a[3].block_hash(), chain_b[3].block_hash());
+
+    let mut state = new_state(Cache::default(), descriptor, genesis);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: chain_a.clone(),
+        spk_hash,
+        txs: Vec::new(),
+        merkle_proof: (Vec::new(), 0),
+    };
+
+    let notify = |chain: &[block::Header]| {
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.headers.subscribe",
+            "params": [{ "hex": serialize_hex(chain.last().expect("non-empty")), "height": 3 }],
+        }))
+    };
+
+    // The A-chain tip. Hold its headers request so the job is still waiting.
+    state.advance(&mut queue, notify(&chain_a))?;
+    let held = queue
+        .drain(..)
+        .filter(|req| req.method.as_ref() == "blockchain.block.headers")
+        .collect::<Vec<_>>();
+    assert!(!held.is_empty(), "the job must have asked for headers");
+    let held_resp = held
+        .iter()
+        .map(|req| response(req, &server))
+        .collect::<Vec<_>>();
+
+    // The server reorgs to B and notifies the new tip at the same height.
+    server.headers = chain_b.clone();
+    state.advance(&mut queue, notify(&chain_b))?;
+    assert!(
+        queue
+            .iter()
+            .any(|req| req.method.as_ref() == "blockchain.block.headers"),
+        "the replacement job must send its own request rather than adopt the one in flight"
+    );
+
+    // The held answer describes the chain the server has left.
+    let mut updates = Vec::new();
+    for resp in held_resp {
+        updates.extend(state.advance(&mut queue, resp)?);
+    }
+    updates.extend(drain_requests(&mut state, &mut queue, &server));
+
+    let tip = updates
+        .iter()
+        .rev()
+        .find_map(|u| u.chain_update.clone())
+        .expect("must get a chain update");
+    assert_eq!(
+        tip.hash(),
+        chain_b[3].block_hash(),
+        "the local chain must end on the tip the server actually has"
+    );
+    Ok(())
+}
+
+/// A headers batch answered from a chain other than the one announced.
+///
+/// `blockchain.block.headers` is answered from whichever chain the server holds when it *replies*,
+/// so a reorg between receiving the request and answering it returns blocks for a tip we were
+/// never told about. Adopting them would put the checkpoint chain on a chain no notification ever
+/// announced, and no notification would arrive to correct it.
+#[test]
+fn headers_for_a_chain_we_were_not_told_about_are_not_adopted() -> anyhow::Result<()> {
+    let (descriptor, spk_hash, _spk) = tracked_descriptor()?;
+    let (genesis, h1) = base_headers();
+    let h2 = block_with_root(&h1, TxMerkleNode::all_zeros(), 200, 0);
+    let a3 = block_with_root(&h2, TxMerkleNode::all_zeros(), 300, 0);
+    let b3 = block_with_root(&h2, TxMerkleNode::all_zeros(), 300, 1);
+    assert_ne!(a3.block_hash(), b3.block_hash());
+
+    let mut state = new_state(Cache::default(), descriptor, genesis);
+    let mut queue = ReqQueue::new();
+    let mut server = Server {
+        headers: vec![genesis, h1, h2],
+        spk_hash,
+        txs: Vec::new(),
+        merkle_proof: (Vec::new(), 0),
+    };
+    state.init(&mut queue);
+    drain_requests(&mut state, &mut queue, &server);
+
+    // A3 is announced, so that is the block the job is created to reach.
+    state.advance(
+        &mut queue,
+        raw_msg(json!({
+            "jsonrpc": "2.0",
+            "method": "blockchain.headers.subscribe",
+            "params": [{ "hex": serialize_hex(&a3), "height": 3 }],
+        })),
+    )?;
+
+    // But by the time the server answers, it is on B3 — and it does not announce it, because
+    // this response is the reorg's only appearance.
+    server.headers = vec![genesis, h1, h2, b3];
+    let updates = drain_requests(&mut state, &mut queue, &server);
+
+    let tip = updates.iter().rev().find_map(|u| u.chain_update.clone());
+    assert_ne!(
+        tip.as_ref().map(|cp| cp.hash()),
+        Some(b3.block_hash()),
+        "a chain no notification announced must not be adopted"
+    );
+    assert_ne!(
+        tip.as_ref().map(|cp| cp.hash()),
+        Some(a3.block_hash()),
+        "and the announced block was never actually delivered"
+    );
+    Ok(())
+}