fix(hotblocks): enforce finalized fork floor

crates/data-client/src/reqwest/client.rs

@@ -36,9 +36,7 @@ pub struct ReqwestDataClient {
 
 impl Debug for ReqwestDataClient {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("ReqwestDataClient")
-            .field("url", &self.url.as_str())
-            .finish()
+        f.write_str(self.url.as_str())
     }
 }
 

crates/data-source/src/standard.rs

@@ -5,7 +5,7 @@ use futures::{future::BoxFuture, stream::BoxStream, FutureExt, Stream, StreamExt
 use sqd_data_client::{BlockStreamRequest, BlockStreamResponse, DataClient};
 use sqd_primitives::{Block, BlockNumber, BlockRef};
 use tokio::time::Sleep;
-use tracing::warn;
+use tracing::{info, warn};
 
 use crate::types::{DataEvent, DataSource};
 
@@ -78,6 +78,14 @@ impl<F> DataSourceState<F> {
                     }
                     Poll::Ready(Ok(BlockStreamResponse::Fork(prev_blocks))) => {
                         ep.error_counter = 0;
+                        info!(
+                            data_source =? ep.client,
+                            stream_from = req.first_block,
+                            hint_count = prev_blocks.len(),
+                            oldest_hint =? prev_blocks.first().map(|b| b.number),
+                            newest_hint =? prev_blocks.last().map(|b| b.number),
+                            "upstream reported a fork"
+                        );
                         ep.state = EndpointState::Fork {
                             req: req.clone(),
                             prev_blocks
@@ -289,11 +297,19 @@ where
 
         let forks = self.endpoints.iter().filter(|ep| ep.is_on_fork()).count();
         if forks > 0 {
-            if forks > self.endpoints.len() / 2
-                || forks == self.endpoints.iter().filter(|ep| ep.is_active()).count()
-                || self.fork_consensus_timeout(cx)
-            {
-                return Poll::Ready(DataEvent::Fork(self.extract_fork()));
+            let active = self.endpoints.iter().filter(|ep| ep.is_active()).count();
+            if forks > self.endpoints.len() / 2 || forks == active || self.fork_consensus_timeout(cx) {
+                let chain = self.extract_fork();
+                info!(
+                    forked_endpoints = forks,
+                    active_endpoints = active,
+                    total_endpoints = self.endpoints.len(),
+                    hint_count = chain.len(),
+                    oldest_hint =? chain.first().map(|b| b.number),
+                    newest_hint =? chain.last().map(|b| b.number),
+                    "fork consensus reached"
+                );
+                return Poll::Ready(DataEvent::Fork(chain));
             }
         } else {
             self.state.fork_consensus_timeout = None

crates/hotblocks-harness/README.md

@@ -22,6 +22,7 @@ is what makes the crash/restart and shutdown classes expressible at all.
 ```bash
 cargo test -p sqd-hotblocks-harness          # the harness's own unit tests (model, chain, simulator)
 cargo test -p sqd-hotblocks --test ct1_happy_path   # CT-1 — the Phase 0 exit criterion
+cargo test -p sqd-hotblocks --test ct4_finality     # CT-4 — finalized-prefix equivocation
 cargo test -p sqd-hotblocks --test ct9_source_faults
 ```
 
@@ -33,7 +34,7 @@ reusable lives here, so a future soak or benchmark runner can use it outside `ca
 
 | Module | What it is | Spec |
 |---|---|---|
-| [`sim`](src/sim.rs) | source simulator: scripted chain, fork signals, finality headers, fault knobs | 13 §7, DEF-12 |
+| [`sim`](src/sim.rs) | source simulator: scripted chain, fork signals, finality headers, fault knobs including explicit finalized-prefix equivocation | 13 §7, DEF-12 |
 | [`model`](src/model.rs) | the reference model — the oracle. Block-exact, well-formedness asserted after every transition | 12 §2 |
 | [`driver`](src/driver.rs) | client: the read binding, the structural validators, the anchored follower and backfill scanner | 04 §7, 12 §4 |
 | [`compare`](src/compare.rs) | quiescence comparator: diffs every observable, collects *all* violations before failing | 12 §1 |
@@ -122,9 +123,12 @@ Fixed in `crates/data-client/src/reqwest/lines.rs`; pinned by a unit test there
 - **CT-2 (crash/restart)** — `Sut::crash()`, `Sut::stop()`, `Sut::restart()` already exist and
   keep the same database directory and port across boots. What is missing is the kill-point
   matrix.
-- **CT-4 (fork/finality corpus)** — `Harness::fork()` and the model's `resolve_fork` /
-  `Finalize::IntegrityFault` are implemented and unit-tested; the follower implements the
-  normative CONFLICT recovery of 04 §7. What is missing is the scripts.
+- **CT-4 (fork/finality corpus)** — `ct4_finality` drives source-equivocation faults through the
+  real binary at both the retained-window floor and a deterministic two-chunk layout with finality
+  strictly inside the second chunk. Both prove rollback resumes at `fin + 1` and the accepted
+  finalized prefix remains unchanged. `Harness::fork()`, the model's `resolve_fork` /
+  `Finalize::IntegrityFault`, and the follower's normative CONFLICT recovery of 04 §7 support the
+  remaining successful-reorg, below-window, malformed-finality, fork-storm and alarm scripts.
 - **CT-5 (error taxonomy)** — `ct5_error_soundness` covers unsupported-dialect containment,
   error classification, and mid-stream worker-panic abort; the anchored check across large
   sparse-number holes is deferred (GAP-21, test `#[ignore]`d). `Model::predict_query` supplies

crates/hotblocks-harness/src/harness.rs

@@ -29,6 +29,8 @@ pub struct HarnessConfig {
     /// Dense (evm, hyperliquid) or sparse (Solana slots) block numbering.
     pub numbering: Numbering,
     pub retention: Retention,
+    /// Keep physical ingest chunks separate when a test needs a deterministic storage layout.
+    pub disable_compaction: bool,
     /// Whether the service is told the anchor hash. If not, the anchor is `⊥` (DEF-7) and the
     /// first block's parent is unverifiable.
     pub anchored: bool,
@@ -54,6 +56,7 @@ impl HarnessConfig {
                 number: start_block,
                 parent_hash: Some(block_hash(start_block - 1, 0))
             },
+            disable_compaction: false,
             anchored: true,
             source_poll: Duration::from_millis(200),
             rust_log: "info".to_string(),
@@ -96,6 +99,7 @@ impl Harness {
                 id: cfg.dataset.clone(),
                 kind: cfg.chain.config_kind().to_string(),
                 retention: cfg.retention.clone(),
+                disable_compaction: cfg.disable_compaction,
                 sources: vec![sim.base_url(&cfg.dataset)]
             }]
         );

crates/hotblocks-harness/src/sim.rs

@@ -87,6 +87,8 @@ pub struct SimFaults {
 #[derive(Clone, Copy, Debug, Default)]
 pub struct SimStats {
     pub stream_requests: u64,
+    /// `fromBlock` on the most recent request, for rollback-position assertions.
+    pub last_stream_from: Option<BlockNumber>,
     pub blocks_served: u64,
     pub fork_signals: u64,
     pub no_data: u64,
@@ -161,6 +163,18 @@ impl SourceSim {
         Ok(blocks)
     }
 
+    /// Fault injection for CT-4/FM-SRC-5: replace a suffix that includes the source's own
+    /// finalized head and claim the replacement tip as final.
+    ///
+    /// Unlike [`Self::fork`], this deliberately violates source finality. It has no model-side
+    /// counterpart: the last accepted model state remains the oracle while the SUT rejects the
+    /// equivocating source.
+    pub fn equivocate_finalized_prefix(&self, dataset: &str, from: BlockNumber, len: u32) -> Result<()> {
+        self.try_with(dataset, |d| d.equivocate_finalized_prefix(from, len))?;
+        self.bump();
+        Ok(())
+    }
+
     /// Declare `number` (and everything below it) final.
     pub fn finalize(&self, dataset: &str, number: BlockNumber) -> Result<BlockRef> {
         let r = self.try_with(dataset, |d| d.finalize(number))?;
@@ -301,22 +315,49 @@ impl DatasetSim {
     }
 
     fn fork(&mut self, from: BlockNumber, len: u32) -> Result<Vec<Block>> {
+        self.validate_fork_position(from)?;
+        ensure!(
+            self.fin.as_ref().is_none_or(|f| f.number < from),
+            "the script forks at or below the source's own finalized head — an equivocating source \
+             belongs to the CT-4 fault corpus, not to a well-formed script"
+        );
+        Ok(self.replace_suffix(from, len))
+    }
+
+    fn equivocate_finalized_prefix(&mut self, from: BlockNumber, len: u32) -> Result<()> {
+        self.validate_fork_position(from)?;
+        ensure!(len > 0, "a finality-equivocation fault must mint a replacement tip");
+        let finalized = self
+            .fin
+            .as_ref()
+            .context("a finality-equivocation fault requires an existing finalized head")?;
+        ensure!(
+            from <= finalized.number,
+            "equivocation at {from} does not replace finalized block {}",
+            finalized.number
+        );
+
+        let replacement = self.replace_suffix(from, len);
+        self.fin = Some(replacement.last().expect("a non-empty replacement has a tip").as_ref());
+        Ok(())
+    }
+
+    fn validate_fork_position(&self, from: BlockNumber) -> Result<()> {
         ensure!(
             from >= self.start,
             "fork at {from} is below the source's first block {}",
             self.start
         );
         ensure!(from <= self.next_number(), "fork at {from} is above the source's chain");
-        ensure!(
-            self.fin.as_ref().is_none_or(|f| f.number < from),
-            "the script forks at or below the source's own finalized head — an equivocating source \
-             belongs to the CT-4 fault corpus, not to a well-formed script"
-        );
+        Ok(())
+    }
+
+    fn replace_suffix(&mut self, from: BlockNumber, len: u32) -> Vec<Block> {
         let keep = self.chain.partition_point(|b| b.number < from);
         self.chain.truncate(keep);
         self.fork_id = self.next_fork_id;
         self.next_fork_id += 1;
-        Ok(self.produce(len))
+        self.produce(len)
     }
 
     fn finalize(&mut self, number: BlockNumber) -> Result<BlockRef> {
@@ -342,6 +383,7 @@ impl DatasetSim {
 
     fn respond(&mut self, req: &StreamReq) -> Reply {
         self.stats.stream_requests += 1;
+        self.stats.last_stream_from = Some(req.from_block);
 
         let parent_pos = req.from_block.saturating_sub(1);
         if let Some(asserted) = &req.parent_block_hash {

crates/hotblocks-harness/src/sut.rs

@@ -39,6 +39,8 @@ pub struct DatasetSpec {
     pub id: String,
     pub kind: String,
     pub retention: Retention,
+    /// Preserve response-aligned chunks for tests that exercise physical layout boundaries.
+    pub disable_compaction: bool,
     pub sources: Vec<String>
 }
 
@@ -235,6 +237,7 @@ impl Sut {
                 Retention::Api => yaml.push_str("  retention_strategy: Api\n"),
                 Retention::None => yaml.push_str("  retention_strategy: None\n")
             }
+            yaml.push_str(&format!("  disable_compaction: {}\n", ds.disable_compaction));
             yaml.push_str("  data_sources:\n");
             for src in &ds.sources {
                 yaml.push_str(&format!("    - \"{src}\"\n"));