diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs
index b6d9da82c..1419df0b6 100644
--- a/data_plane/src/drivers/ingest/otel.rs
+++ b/data_plane/src/drivers/ingest/otel.rs
@@ -604,7 +604,8 @@ fn resolve_bucket_sid_for_agg_config(
     ingest_state: &Arc<IngestState>,
     config: &asap_types::aggregation_config::AggregationConfig,
     point_labels: &HashMap<String, String>,
-) -> (u64, asap_types::PolicyFingerprint) {
+    captured_generation: Option<&asap_types::sds::CatalogGeneration>,
+) -> Result<(u64, asap_types::PolicyFingerprint), String> {
     let grouping_pairs: Vec<(&str, &str)> = config
         .grouping_labels
         .labels
@@ -617,11 +618,28 @@ fn resolve_bucket_sid_for_agg_config(
     let fp = crate::drivers::ingest::canonical_attrs_fingerprint(&grouping_pairs);
     let agg_kind_canonical =
         crate::storage_engines::sketch_db::data::materialization_kind_for_config(config);
-    let sid = ingest_state
-        .series_resolver
-        .resolve(&config.metric, &fp, &agg_kind_canonical);
+    let sid = ingest_state.series_resolver.resolve_with_reactivation(
+        &config.metric,
+        &fp,
+        &agg_kind_canonical,
+        |sid| {
+            ingest_state
+                .sketch_index
+                .validate_routed_catalog_generation(captured_generation)?;
+            let activation = ingest_state
+                .sketch_index
+                .authorize_series_reactivation(sid, config.policy_fingerprint().into())?;
+            if activation
+                .as_deref()
+                .is_some_and(|generation| Some(generation) != captured_generation)
+            {
+                return Err("stale OTLP generation cannot reactivate series".into());
+            }
+            Ok(activation)
+        },
+    )?;
     let policy_fp = asap_types::PolicyFingerprint(config.policy_fp_u64());
-    (sid, policy_fp)
+    Ok((sid, policy_fp))
 }
 
 async fn route_otlp_to_precompute(
@@ -633,7 +651,15 @@ async fn route_otlp_to_precompute(
 
     // Snapshot the latest agg_configs from the hot-reload handle so
     // new aggregations are visible without restart.
-    let snap = ingest_state.config_snapshot();
+    let physical_plan_snapshot = ingest_state.physical_plan_snapshot();
+    let catalog_generation = physical_plan_snapshot
+        .as_ref()
+        .and_then(|plan| plan.precompute_plan.summary_catalog.clone())
+        .map(Arc::new);
+    let snap = physical_plan_snapshot
+        .as_ref()
+        .map(|plan| plan.runtime_config.clone())
+        .unwrap_or_else(|| ingest_state.config_snapshot());
     let agg_configs = snap.get_all_aggregation_configs();
     // Schema retirement #5 — the agg_id-keyed `SchemaRegistry` is
     // gone; sid-level lifecycle now lives on `SketchStore`. Reconcile
@@ -691,8 +717,18 @@ async fn route_otlp_to_precompute(
                 continue;
             }
             let group_key = IngestState::extract_group_key_for(&series_key, config);
-            let (sid, policy_fp) =
-                resolve_bucket_sid_for_agg_config(ingest_state, config, &point.labels);
+            let (sid, policy_fp) = match resolve_bucket_sid_for_agg_config(
+                ingest_state,
+                config,
+                &point.labels,
+                catalog_generation.as_deref(),
+            ) {
+                Ok(binding) => binding,
+                Err(error) => {
+                    warn!(%error, "configured ingest series reactivation rejected");
+                    continue;
+                }
+            };
             by_bucket
                 .entry(sid)
                 .or_insert_with(|| ((sid, policy_fp, group_key.clone()), Vec::new()))
@@ -724,7 +760,7 @@ async fn route_otlp_to_precompute(
     if !raw_messages.is_empty() {
         if let Err(e) = ingest_state
             .router
-            .route_group_batch(raw_messages, ingest_received_at)
+            .route_group_batch(raw_messages, ingest_received_at, catalog_generation.clone())
             .await
         {
             warn!("OTLP raw-sample routing error: {}", e);
@@ -782,8 +818,18 @@ async fn route_otlp_to_precompute(
             // `reconcile_from_streaming_config` derives from the same
             // config (otherwise the bucket would be reachable but never
             // reconciled).
-            let (sid, policy_fp) =
-                resolve_bucket_sid_for_agg_config(ingest_state, config, &point.labels);
+            let (sid, policy_fp) = match resolve_bucket_sid_for_agg_config(
+                ingest_state,
+                config,
+                &point.labels,
+                catalog_generation.as_deref(),
+            ) {
+                Ok(binding) => binding,
+                Err(error) => {
+                    warn!(%error, "configured ingest series reactivation rejected");
+                    continue;
+                }
+            };
             sketch_messages.push(WorkerMessage::AccumulatorInput {
                 sid,
                 policy_fp,
@@ -812,7 +858,11 @@ async fn route_otlp_to_precompute(
     if !sketch_messages.is_empty() {
         if let Err(e) = ingest_state
             .router
-            .route_group_batch(sketch_messages, ingest_received_at)
+            .route_group_batch(
+                sketch_messages,
+                ingest_received_at,
+                catalog_generation.clone(),
+            )
             .await
         {
             warn!("OTLP sketch routing error: {}", e);
@@ -875,6 +925,10 @@ async fn route_modified_otlp_sketches_to_precompute(
         .as_ref()
         .map(|plan| plan.runtime_config.clone())
         .unwrap_or_else(|| ingest_state.config_snapshot());
+    let catalog_generation = physical_plan_snapshot
+        .as_ref()
+        .and_then(|plan| plan.precompute_plan.summary_catalog.clone())
+        .map(Arc::new);
     let active_physical_plan = physical_plan_snapshot.filter(|plan| plan.plan_id() != 0);
     let lineage_batch_guard = active_physical_plan
         .as_ref()
@@ -1219,11 +1273,42 @@ async fn route_modified_otlp_sketches_to_precompute(
                             spatial_filter_canonical: String::new(),
                         };
                         let agg_kind_canonical = agg_kind.canonical_string();
-                        let assigned = ingest_state.series_resolver.resolve(
+                        let definition = frame_identity
+                            .as_ref()
+                            .map(|frame| frame.materialization)
+                            .unwrap_or_else(|| asap_types::PolicyFingerprint(0).into());
+                        let assigned = match ingest_state.series_resolver.resolve_with_reactivation(
                             &canonical_name,
                             &fp,
                             &agg_kind_canonical,
-                        );
+                            |sid| {
+                                ingest_state
+                                    .sketch_index
+                                    .validate_routed_catalog_generation(
+                                        catalog_generation.as_deref(),
+                                    )?;
+                                let activation = ingest_state
+                                    .sketch_index
+                                    .authorize_series_reactivation(sid, definition)?;
+                                if activation.as_deref().is_some_and(|generation| {
+                                    Some(generation) != catalog_generation.as_deref()
+                                }) {
+                                    return Err(
+                                        "stale OTLP generation cannot reactivate series".into()
+                                    );
+                                }
+                                Ok(activation)
+                            },
+                        ) {
+                            Ok(sid) => sid,
+                            Err(error) => {
+                                if dp.series_id != 0 {
+                                    unknown_sids.push(dp.series_id);
+                                }
+                                warn!(%error, "modified OTLP series reactivation rejected");
+                                continue;
+                            }
+                        };
                         if dp.series_id != 0 && dp.series_id != assigned {
                             // Sender's cached sid disagrees with the
                             // resolver's binding — sender's cache is
@@ -1751,8 +1836,18 @@ async fn route_modified_otlp_sketches_to_precompute(
                             // PERF-3 — `dp.attrs` is already a
                             // `HashMap<String, String>`; pass it directly
                             // instead of rebuilding `attrs_map` per config.
-                            let (bucket_sid, policy_fp) =
-                                resolve_bucket_sid_for_agg_config(ingest_state, config, &dp.attrs);
+                            let (bucket_sid, policy_fp) = match resolve_bucket_sid_for_agg_config(
+                                ingest_state,
+                                config,
+                                &dp.attrs,
+                                catalog_generation.as_deref(),
+                            ) {
+                                Ok(binding) => binding,
+                                Err(error) => {
+                                    warn!(%error, "configured ingest series reactivation rejected");
+                                    continue;
+                                }
+                            };
                             let acc_for_msg = if i + 1 == n {
                                 // Last (or only) match — move the owned
                                 // accumulator out, no clone.
@@ -1803,7 +1898,7 @@ async fn route_modified_otlp_sketches_to_precompute(
     if !messages.is_empty() {
         if let Err(e) = ingest_state
             .router
-            .route_group_batch(messages, ingest_received_at)
+            .route_group_batch(messages, ingest_received_at, catalog_generation.clone())
             .await
         {
             warn!("OTLP modified-proto sketch routing error: {}", e);
diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
index d557031d9..367ac15f1 100644
--- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs
+++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
@@ -142,6 +142,8 @@ pub enum RemoteWriteError {
     DedupCapacity(usize),
     #[error("Remote Write requires an active PrometheusRemoteWriteV1 PhysicalPlan")]
     InactivePhysicalPlan,
+    #[error("series identity admission failed: {0}")]
+    SeriesIdentity(String),
     #[error(transparent)]
     Backpressure(#[from] TryRouteError),
 }
@@ -331,7 +333,7 @@ impl PrometheusRemoteWriteReceiver {
             return Err(RemoteWriteError::DedupCapacity(config.max_dedup_entries));
         }
 
-        let messages = route_messages(&new_samples, &self.inner.ingest, &physical_plan);
+        let messages = route_messages(&new_samples, &self.inner.ingest, &physical_plan)?;
         let generation = Arc::new(
             physical_plan
                 .precompute_plan
@@ -618,7 +620,7 @@ fn route_messages(
     samples: &[CanonicalSample],
     ingest: &Arc<IngestState>,
     physical_plan: &crate::storage_engines::types::ActivePhysicalPlan,
-) -> Vec<WorkerMessage> {
+) -> Result<Vec<WorkerMessage>, RemoteWriteError> {
     type Bucket = (
         u64,
         asap_types::PolicyFingerprint,
@@ -706,10 +708,25 @@ fn route_messages(
             // value-weighted Top-K). Keep those states on distinct SIDs.
             let materialization_kind =
                 crate::storage_engines::sketch_db::data::materialization_kind_for_config(config);
-            let sid =
-                ingest
-                    .series_resolver
-                    .resolve(&config.metric, attrs_fp, &materialization_kind);
+            let sid = ingest
+                .series_resolver
+                .resolve_with_reactivation(&config.metric, attrs_fp, &materialization_kind, |sid| {
+                    ingest.sketch_index.validate_routed_catalog_generation(
+                        physical_plan.precompute_plan.summary_catalog.as_ref(),
+                    )?;
+                    let activation = ingest
+                        .sketch_index
+                        .authorize_series_reactivation(sid, policy_fp.into())?;
+                    if let Some(generation) = &activation {
+                        if physical_plan.precompute_plan.summary_catalog.as_ref()
+                            != Some(generation.as_ref())
+                        {
+                            return Err("stale routed generation cannot reactivate series".into());
+                        }
+                    }
+                    Ok(activation)
+                })
+                .map_err(RemoteWriteError::SeriesIdentity)?;
             buckets
                 .entry(sid)
                 .or_insert_with(|| ((sid, policy_fp, group_key), Vec::new()))
@@ -718,7 +735,7 @@ fn route_messages(
         }
     }
     let received_at = Instant::now();
-    buckets
+    Ok(buckets
         .into_values()
         .map(
             |((sid, policy_fp, group_key), samples)| WorkerMessage::GroupSamples {
@@ -729,7 +746,7 @@ fn route_messages(
                 ingest_received_at: received_at,
             },
         )
-        .collect()
+        .collect())
 }
 
 #[derive(Debug)]
@@ -1082,6 +1099,10 @@ mod tests {
             sketch_index: Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()),
             observability: IngestObservability::default(),
         });
+        ingest
+            .sketch_index
+            .install_summary_catalog(physical_plan.summary_catalog.as_ref().unwrap().clone())
+            .unwrap();
         let request = WriteRequest {
             timeseries: ["api", "order", "payment", "user", "webapp"]
                 .into_iter()
@@ -1113,7 +1134,7 @@ mod tests {
         };
         let samples = canonicalize_request(&request, &PrometheusRemoteWriteConfig::default())
             .expect("canonical samples");
-        let messages = route_messages(&samples, &ingest, &physical_plan);
+        let messages = route_messages(&samples, &ingest, &physical_plan).unwrap();
         let mut cms_buckets = 0;
         let mut counter_buckets = 0;
         let mut cms_samples = 0;
@@ -1204,7 +1225,7 @@ mod tests {
         let drain = tokio::spawn(async move { handle.drain().await });
         assert!(matches!(
             worker.recv().await.unwrap(),
-            WorkerMessage::Admitted { .. }
+            WorkerMessage::BoundInput { .. }
         ));
         let WorkerMessage::Drain(reply) = worker.recv().await.unwrap() else {
             panic!("expected barrier")
@@ -1506,11 +1527,15 @@ mod tests {
         let (receiver, mut worker) = configured_receiver();
         receiver.accept(&one_sample(4.0)).unwrap();
         let message = worker.recv().await.expect("routed worker message");
-        let WorkerMessage::Admitted { input, revision } = message else {
+        let WorkerMessage::BoundInput {
+            input, revision, ..
+        } = message
+        else {
             panic!("missing admission receipt")
         };
-        assert!(revision.revision > 0);
+        assert!(revision.as_ref().unwrap().revision > 0);
         let message = *input;
+        let revision = revision.expect("remote-write input carries admission receipt");
         let WorkerMessage::GroupSamples {
             group_key, samples, ..
         } = message
diff --git a/data_plane/src/drivers/ingest/series_resolver.rs b/data_plane/src/drivers/ingest/series_resolver.rs
index 121bd3c5d..c7d6389b6 100644
--- a/data_plane/src/drivers/ingest/series_resolver.rs
+++ b/data_plane/src/drivers/ingest/series_resolver.rs
@@ -25,9 +25,10 @@
 //! in `docs/design_docs/series-identity.md`.
 
 use dashmap::DashMap;
+use std::collections::BTreeMap;
 use std::fs::{File, OpenOptions};
 use std::io::{Read, Seek, SeekFrom, Write};
-use std::path::{Path, PathBuf};
+use std::path::PathBuf;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use tracing::{info, warn};
@@ -133,49 +134,110 @@ impl SeriesIdResolver {
     /// stale via `unknown_series_ids` and re-resolve with attrs.
     ///
     /// Persistence failures are logged at WARN and do NOT propagate —
-    /// the resolver stays in-memory-correct. Next restart will not
-    /// recover the lost mint, and the agent will hit the eviction
-    /// recovery path (one extra round trip with attrs).
+    /// the returned identity is ephemeral and is not cached. Catalog-bound
+    /// callers use `try_resolve` and receive the persistence failure instead.
     pub fn resolve(
         &self,
         metric_name: &str,
         attrs_fingerprint: &str,
         agg_kind_canonical: &str,
     ) -> u64 {
+        match self.try_resolve(metric_name, attrs_fingerprint, agg_kind_canonical) {
+            Ok(sid) => sid,
+            Err(error) => {
+                // Compatibility callers still receive an ephemeral ID, but it
+                // must never enter the shared cache used by strict producers.
+                let sid = self.next_sid.fetch_add(1, Ordering::Relaxed);
+                warn!(%error, sid, "resolver persistence failed; returning uncached ephemeral identity");
+                sid
+            }
+        }
+    }
+
+    /// Persist a new binding before exposing it to catalog-bound producers.
+    pub fn try_resolve(&self, metric: &str, attrs: &str, kind: &str) -> std::io::Result<u64> {
+        use dashmap::mapref::entry::Entry;
+        let key = (metric.to_owned(), attrs.to_owned(), kind.to_owned());
+        match self.cache.entry(key) {
+            Entry::Occupied(entry) => Ok(*entry.get()),
+            Entry::Vacant(entry) => {
+                let sid = self
+                    .next_sid
+                    .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| {
+                        value.checked_add(1)
+                    })
+                    .map_err(|_| std::io::Error::other("series ID exhausted"))?;
+                self.persistence.append(sid, metric, attrs, kind)?;
+                entry.insert(sid);
+                Ok(sid)
+            }
+        }
+    }
+
+    /// Resolve a logical series and ask the storage lifecycle owner whether
+    /// an explicit catalog activation requires a fresh physical lifetime.
+    pub fn resolve_with_reactivation(
+        &self,
+        metric: &str,
+        attrs: &str,
+        kind: &str,
+        authorize: impl FnOnce(u64) -> Result<Option<Arc<asap_types::sds::CatalogGeneration>>, String>,
+    ) -> Result<u64, String> {
+        let sid = self
+            .try_resolve(metric, attrs, kind)
+            .map_err(|error| error.to_string())?;
+        match authorize(sid)? {
+            None => Ok(sid),
+            Some(generation) => self
+                .rotate_for_catalog_activation(metric, attrs, kind, sid, &generation)
+                .map_err(|error| error.to_string()),
+        }
+    }
+
+    /// Advance a tombstoned physical series after the storage engine has
+    /// authorized reactivation in a different installed catalog generation.
+    /// The logical cache key remains unchanged. Persistence failure leaves
+    /// the old binding intact. Concurrent activation requires fresh authorization.
+    pub fn rotate_for_catalog_activation(
+        &self,
+        metric_name: &str,
+        attrs_fingerprint: &str,
+        agg_kind_canonical: &str,
+        previous_sid: u64,
+        generation: &asap_types::sds::CatalogGeneration,
+    ) -> std::io::Result<u64> {
         let key = (
-            metric_name.to_string(),
-            attrs_fingerprint.to_string(),
-            agg_kind_canonical.to_string(),
+            metric_name.to_owned(),
+            attrs_fingerprint.to_owned(),
+            agg_kind_canonical.to_owned(),
         );
-        // Fast path: read-only check on the cache before taking the
-        // bucket's write lock. DashMap's `get` takes a shard read lock;
-        // the common case (a hit on a known identity) never serializes
-        // against other resolve calls.
-        if let Some(existing) = self.cache.get(&key) {
-            return *existing;
+        let mut binding = self.cache.get_mut(&key).ok_or_else(|| {
+            std::io::Error::new(
+                std::io::ErrorKind::NotFound,
+                "catalog activation requires an existing resolver binding",
+            )
+        })?;
+        if *binding != previous_sid {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::WouldBlock,
+                "physical series changed after catalog authorization; retry routing",
+            ));
         }
-        // Slow path: bucket write lock + mint + persist + insert.
-        // `entry().or_insert_with` ensures only ONE caller runs the
-        // closure for a given key, even under concurrent load. The
-        // persistence append happens inside the closure so the binding
-        // is durable before any caller observes the sid.
-        let entry = self.cache.entry(key).or_insert_with(|| {
-            let sid = self.next_sid.fetch_add(1, Ordering::Relaxed);
-            if let Err(e) =
-                self.persistence
-                    .append(sid, metric_name, attrs_fingerprint, agg_kind_canonical)
-            {
-                warn!(
-                    metric = %metric_name,
-                    sid,
-                    error = %e,
-                    "resolver persistence append failed; binding is \
-                     in-memory-only and will not survive restart",
-                );
-            }
-            sid
-        });
-        *entry
+        let replacement = self
+            .next_sid
+            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |sid| {
+                sid.checked_add(1)
+            })
+            .map_err(|_| std::io::Error::other("physical series ID space exhausted"))?;
+        self.persistence.append_catalog_activation(
+            replacement,
+            metric_name,
+            attrs_fingerprint,
+            agg_kind_canonical,
+            generation,
+        )?;
+        *binding = replacement;
+        Ok(replacement)
     }
 
     /// Look up an existing sid without minting. Returns `None` if the
@@ -221,34 +283,17 @@ impl Default for SeriesIdResolver {
 // writes a WAL record per fresh mint; the resolver replays it on startup
 // so the agent's cached sids stay valid across backend restarts.
 //
-// WAL format v3 (current; v1 was attrs-only, never shipped to prod;
-// v2 used `agg_kind_canonical` without spatial-filter, replaced when
-// spatial_filter_canonical was folded into agg_kind_canonical to
-// distinguish filter-distinct policies on the sid identity):
-//   header: 8 bytes  → b"ASAPSRP\x03"
-//   record: 8 bytes  → sid (u64 little-endian)
-//           4 bytes  → metric_len (u32 LE)
-//           metric_len bytes → metric utf8
-//           4 bytes  → fp_len (u32 LE)
-//           fp_len bytes → fp utf8
-//           4 bytes  → agg_kind_len (u32 LE)
-//           agg_kind_len bytes → agg_kind_canonical utf8
-//
-// v2 WALs are not auto-migrated — pre-prod constraint. A v2 header
-// causes `FilePersistence::open` to fail with `InvalidData`; recovery
-// is to delete the file and let the resolver cold-start (the
-// `unknown_series_ids` eviction primitive handles the bandwidth blip).
+// WAL v4: header b"ASAPSRP\x04", followed by u32-length-prefixed JSON
+// records. CatalogGeneration records are normalized by snapshot digest;
+// Binding records carry a reference only for explicitly authorized rotations.
+// Opening a v3 WAL atomically migrates its durable prefix without changing
+// physical IDs. v1/v2 remain unsupported because they used different identity
+// semantics. Older binaries reject the v4 header rather than truncating it.
 //
-// Append-only; sids are minted once and never rewritten, so the log size
-// is proportional to live cardinality. At 100M sids (~5GB) compaction
-// becomes worth scheduling; not implemented here.
-//
-// Crash safety: every `append` calls `fsync` before returning. A torn
-// write at EOF (kernel buffered the bytes but the metadata flush was
-// interrupted) is detected at replay via short-read on any record field
-// — the file is truncated to the last durable record's offset and replay
-// returns the durable prefix. No CRC: bit-rot is low-probability for an
-// append-only WAL; add a CRC field if telemetry ever shows it firing.
+// Every append is fsynced before publication. Failed appends roll back to the
+// previous offset; replay truncates incomplete trailing frames and rejects
+// malformed or oversized complete frames. The append-only WAL has no garbage
+// collection yet; lifetime rotations add bindings for the same logical key.
 
 /// One durable binding row read back from the WAL.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -257,6 +302,7 @@ pub struct ResolverRecord {
     pub metric: String,
     pub attrs_fingerprint: String,
     pub agg_kind_canonical: String,
+    pub catalog_generation: Option<Arc<asap_types::sds::CatalogGeneration>>,
 }
 
 /// Durability hook for [`SeriesIdResolver`]. Implementations decide
@@ -277,6 +323,23 @@ pub trait SeriesResolverPersistence: Send + Sync {
         agg_kind_canonical: &str,
     ) -> std::io::Result<()>;
 
+    /// Persist a replacement physical lifetime with its catalog provenance.
+    /// Backends must explicitly implement this stronger contract; silently
+    /// downgrading to a legacy binding append would lose authorization.
+    fn append_catalog_activation(
+        &self,
+        _sid: u64,
+        _metric: &str,
+        _attrs_fingerprint: &str,
+        _agg_kind_canonical: &str,
+        _generation: &asap_types::sds::CatalogGeneration,
+    ) -> std::io::Result<()> {
+        Err(std::io::Error::new(
+            std::io::ErrorKind::Unsupported,
+            "resolver persistence does not support catalog activation",
+        ))
+    }
+
     /// Read every durable binding in append order. Called once at
     /// resolver construction time.
     fn replay(&self) -> std::io::Result<Vec<ResolverRecord>>;
@@ -298,6 +361,17 @@ impl SeriesResolverPersistence for NoopPersistence {
         Ok(())
     }
 
+    fn append_catalog_activation(
+        &self,
+        _sid: u64,
+        _metric: &str,
+        _attrs_fingerprint: &str,
+        _agg_kind_canonical: &str,
+        _generation: &asap_types::sds::CatalogGeneration,
+    ) -> std::io::Result<()> {
+        Ok(())
+    }
+
     fn replay(&self) -> std::io::Result<Vec<ResolverRecord>> {
         Ok(Vec::new())
     }
@@ -309,10 +383,40 @@ impl SeriesResolverPersistence for NoopPersistence {
 #[derive(Debug)]
 pub struct FilePersistence {
     file: Mutex<File>,
-    path: PathBuf,
+    generations: Mutex<BTreeMap<String, Arc<asap_types::sds::CatalogGeneration>>>,
+}
+
+const WAL_MAGIC: &[u8; 8] = b"ASAPSRP\x04";
+const LEGACY_WAL_MAGIC: &[u8; 8] = b"ASAPSRP\x03";
+const MAX_WAL_RECORD_BYTES: usize = 4 * 1024 * 1024;
+
+#[derive(serde::Serialize, serde::Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+enum WalRecord {
+    CatalogGeneration {
+        generation: Arc<asap_types::sds::CatalogGeneration>,
+    },
+    Binding {
+        sid: u64,
+        metric: String,
+        attrs_fingerprint: String,
+        agg_kind_canonical: String,
+        #[serde(default)]
+        generation_sha256: Option<String>,
+    },
 }
 
-const WAL_MAGIC: &[u8; 8] = b"ASAPSRP\x03";
+fn write_wal_record(file: &mut File, record: &WalRecord) -> std::io::Result<()> {
+    let bytes = serde_json::to_vec(record).map_err(std::io::Error::other)?;
+    if bytes.len() > MAX_WAL_RECORD_BYTES {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidInput,
+            "resolver WAL record exceeds size limit",
+        ));
+    }
+    file.write_all(&(bytes.len() as u32).to_le_bytes())?;
+    file.write_all(&bytes)
+}
 /// Reject any single field whose length-prefix exceeds these caps. A
 /// corrupted file might claim huge field lengths; without these bounds
 /// the replay loop could allocate gigabytes of zeros before discovering
@@ -324,9 +428,6 @@ const MAX_FP_LEN: usize = 64 * 1024;
 const MAX_AGG_KIND_LEN: usize = 4 * 1024;
 
 impl FilePersistence {
-    /// Open or create the WAL at `path`. On a fresh file, writes the
-    /// magic header and fsyncs. On an existing file, verifies the
-    /// header matches and seeks to EOF for future appends.
     pub fn open(path: PathBuf) -> std::io::Result<Self> {
         if let Some(parent) = path.parent() {
             std::fs::create_dir_all(parent)?;
@@ -336,36 +437,125 @@ impl FilePersistence {
             .write(true)
             .create(true)
             .open(&path)?;
-        let len = file.metadata()?.len();
-        if len == 0 {
+        if file.metadata()?.len() == 0 {
             file.write_all(WAL_MAGIC)?;
             file.sync_all()?;
         } else {
-            let mut hdr = [0u8; 8];
-            file.seek(SeekFrom::Start(0))?;
-            file.read_exact(&mut hdr)?;
-            if &hdr != WAL_MAGIC {
+            let mut header = [0; 8];
+            file.read_exact(&mut header)?;
+            if &header == LEGACY_WAL_MAGIC {
+                let temporary = path.with_extension("v4.tmp");
+                let mut migrated = File::create(&temporary)?;
+                migrated.write_all(WAL_MAGIC)?;
+                loop {
+                    match read_one_record(&mut file) {
+                        ReadOne::Ok(record, offset) => {
+                            debug_assert!(offset >= 8);
+                            write_wal_record(
+                                &mut migrated,
+                                &WalRecord::Binding {
+                                    sid: record.sid,
+                                    metric: record.metric,
+                                    attrs_fingerprint: record.attrs_fingerprint,
+                                    agg_kind_canonical: record.agg_kind_canonical,
+                                    generation_sha256: None,
+                                },
+                            )?;
+                        }
+                        ReadOne::Eof | ReadOne::Torn => break,
+                        ReadOne::Io(error) => return Err(error),
+                        ReadOne::Corrupt => {
+                            return Err(std::io::Error::new(
+                                std::io::ErrorKind::InvalidData,
+                                "corrupt legacy resolver record",
+                            ))
+                        }
+                    }
+                }
+                migrated.sync_all()?;
+                std::fs::rename(&temporary, &path)?;
+                if let Some(parent) = path.parent() {
+                    File::open(parent)?.sync_all()?;
+                }
+                file = OpenOptions::new().read(true).write(true).open(&path)?;
+            } else if &header != WAL_MAGIC {
                 return Err(std::io::Error::new(
                     std::io::ErrorKind::InvalidData,
-                    format!(
-                        "resolver WAL header mismatch at {:?}: expected {:?}, got {:?}",
-                        path, WAL_MAGIC, hdr,
-                    ),
+                    "resolver WAL header mismatch",
                 ));
             }
         }
-        // Position at EOF — appends start here.
         file.seek(SeekFrom::End(0))?;
         Ok(Self {
             file: Mutex::new(file),
-            path,
+            generations: Mutex::new(BTreeMap::new()),
         })
     }
 
-    /// Diagnostic accessor — the WAL path. Tests use this to inspect
-    /// the on-disk file.
-    pub fn path(&self) -> &Path {
-        &self.path
+    fn append_binding(
+        &self,
+        sid: u64,
+        metric: &str,
+        attrs_fingerprint: &str,
+        agg_kind_canonical: &str,
+        generation: Option<&asap_types::sds::CatalogGeneration>,
+    ) -> std::io::Result<()> {
+        if sid == 0
+            || metric.len() > MAX_METRIC_LEN
+            || attrs_fingerprint.len() > MAX_FP_LEN
+            || agg_kind_canonical.len() > MAX_AGG_KIND_LEN
+        {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                "invalid resolver binding",
+            ));
+        }
+        let mut file = self.file.lock().unwrap();
+        let mut generations = self.generations.lock().unwrap();
+        let offset = file.stream_position()?;
+        let result = (|| {
+            if let Some(generation) = generation {
+                if let Some(existing) = generations.get(&generation.snapshot_sha256) {
+                    if existing.as_ref() != generation {
+                        return Err(std::io::Error::new(
+                            std::io::ErrorKind::InvalidInput,
+                            "conflicting catalog generation digest",
+                        ));
+                    }
+                } else {
+                    write_wal_record(
+                        &mut file,
+                        &WalRecord::CatalogGeneration {
+                            generation: Arc::new(generation.clone()),
+                        },
+                    )?;
+                }
+            }
+            write_wal_record(
+                &mut file,
+                &WalRecord::Binding {
+                    sid,
+                    metric: metric.into(),
+                    attrs_fingerprint: attrs_fingerprint.into(),
+                    agg_kind_canonical: agg_kind_canonical.into(),
+                    generation_sha256: generation.map(|value| value.snapshot_sha256.clone()),
+                },
+            )?;
+            file.sync_all()
+        })();
+        if result.is_err() {
+            // Do not append behind an incomplete record after a failed write.
+            file.set_len(offset)?;
+            file.seek(SeekFrom::Start(offset))?;
+            file.sync_all()?;
+            return result;
+        }
+        if let Some(generation) = generation {
+            generations
+                .entry(generation.snapshot_sha256.clone())
+                .or_insert_with(|| Arc::new(generation.clone()));
+        }
+        Ok(())
     }
 }
 
@@ -374,103 +564,157 @@ impl SeriesResolverPersistence for FilePersistence {
         &self,
         sid: u64,
         metric: &str,
-        fp: &str,
+        attrs_fingerprint: &str,
         agg_kind_canonical: &str,
     ) -> std::io::Result<()> {
-        let metric_bytes = metric.as_bytes();
-        let fp_bytes = fp.as_bytes();
-        let agg_kind_bytes = agg_kind_canonical.as_bytes();
-        let metric_len: u32 = metric_bytes.len().try_into().map_err(|_| {
-            std::io::Error::new(
-                std::io::ErrorKind::InvalidInput,
-                "metric name longer than u32::MAX bytes",
-            )
-        })?;
-        let fp_len: u32 = fp_bytes.len().try_into().map_err(|_| {
-            std::io::Error::new(
-                std::io::ErrorKind::InvalidInput,
-                "fingerprint longer than u32::MAX bytes",
-            )
-        })?;
-        let agg_kind_len: u32 = agg_kind_bytes.len().try_into().map_err(|_| {
-            std::io::Error::new(
-                std::io::ErrorKind::InvalidInput,
-                "agg_kind_canonical longer than u32::MAX bytes",
-            )
-        })?;
+        self.append_binding(sid, metric, attrs_fingerprint, agg_kind_canonical, None)
+    }
 
-        let mut f = self.file.lock().unwrap();
-        f.write_all(&sid.to_le_bytes())?;
-        f.write_all(&metric_len.to_le_bytes())?;
-        f.write_all(metric_bytes)?;
-        f.write_all(&fp_len.to_le_bytes())?;
-        f.write_all(fp_bytes)?;
-        f.write_all(&agg_kind_len.to_le_bytes())?;
-        f.write_all(agg_kind_bytes)?;
-        // Durability barrier: caller must not observe the sid until the
-        // record is on stable storage. fsync is the slow part of the
-        // mint path (a few ms on SSD) but it's amortized — minting is
-        // once per identity, not per emit.
-        f.sync_all()?;
-        Ok(())
+    fn append_catalog_activation(
+        &self,
+        sid: u64,
+        metric: &str,
+        attrs_fingerprint: &str,
+        agg_kind_canonical: &str,
+        generation: &asap_types::sds::CatalogGeneration,
+    ) -> std::io::Result<()> {
+        self.append_binding(
+            sid,
+            metric,
+            attrs_fingerprint,
+            agg_kind_canonical,
+            Some(generation),
+        )
     }
 
     fn replay(&self) -> std::io::Result<Vec<ResolverRecord>> {
-        let mut f = self.file.lock().unwrap();
-        f.seek(SeekFrom::Start(0))?;
-        let mut hdr = [0u8; 8];
-        match f.read_exact(&mut hdr) {
-            Ok(()) => {}
-            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
-                // File exists but is empty — caller likely opened it
-                // moments ago without writing the header yet. Treat as
-                // no records.
-                return Ok(Vec::new());
-            }
-            Err(e) => return Err(e),
-        }
-        if &hdr != WAL_MAGIC {
+        let mut file = self.file.lock().unwrap();
+        file.seek(SeekFrom::Start(0))?;
+        let mut header = [0; 8];
+        file.read_exact(&mut header)?;
+        if &header != WAL_MAGIC {
             return Err(std::io::Error::new(
                 std::io::ErrorKind::InvalidData,
-                "resolver WAL header mismatch during replay",
+                "resolver WAL header mismatch",
             ));
         }
-        let mut out = Vec::new();
-        // After successful header read, the offset is 8.
-        let mut safe_offset: u64 = 8;
+        let mut generations = BTreeMap::new();
+        let mut records = Vec::new();
+        let mut physical_keys = BTreeMap::new();
+        let mut logical_sids = BTreeMap::new();
+        let mut safe_offset = 8;
         loop {
-            match read_one_record(&mut *f) {
-                ReadOne::Ok(record, new_offset) => {
-                    out.push(record);
-                    safe_offset = new_offset;
+            let mut length = [0; 4];
+            match file.read_exact(&mut length) {
+                Ok(()) => {}
+                Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => {
+                    file.set_len(safe_offset)?;
+                    break;
                 }
-                ReadOne::Eof => break,
-                ReadOne::Torn => {
-                    warn!(
-                        path = %self.path.display(),
-                        torn_at = safe_offset,
-                        recovered = out.len(),
-                        "resolver WAL: torn record at EOF — truncating to last durable offset",
+                Err(error) => return Err(error),
+            }
+            let length = u32::from_le_bytes(length) as usize;
+            if length > MAX_WAL_RECORD_BYTES {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::InvalidData,
+                    "resolver WAL record exceeds size limit",
+                ));
+            }
+            let mut bytes = vec![0; length];
+            if let Err(error) = file.read_exact(&mut bytes) {
+                if error.kind() == std::io::ErrorKind::UnexpectedEof {
+                    file.set_len(safe_offset)?;
+                    break;
+                }
+                return Err(error);
+            }
+            let record: WalRecord = serde_json::from_slice(&bytes)
+                .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
+            match record {
+                WalRecord::CatalogGeneration { generation } => {
+                    let key = generation.snapshot_sha256.clone();
+                    if generations
+                        .get(&key)
+                        .is_some_and(|existing| existing != &generation)
+                    {
+                        return Err(std::io::Error::new(
+                            std::io::ErrorKind::InvalidData,
+                            "conflicting catalog generation digest",
+                        ));
+                    }
+                    generations.insert(key, generation);
+                }
+                WalRecord::Binding {
+                    sid,
+                    metric,
+                    attrs_fingerprint,
+                    agg_kind_canonical,
+                    generation_sha256,
+                } => {
+                    if sid == 0
+                        || metric.len() > MAX_METRIC_LEN
+                        || attrs_fingerprint.len() > MAX_FP_LEN
+                        || agg_kind_canonical.len() > MAX_AGG_KIND_LEN
+                    {
+                        return Err(std::io::Error::new(
+                            std::io::ErrorKind::InvalidData,
+                            "invalid resolver binding",
+                        ));
+                    }
+                    let key = (
+                        metric.clone(),
+                        attrs_fingerprint.clone(),
+                        agg_kind_canonical.clone(),
                     );
-                    f.set_len(safe_offset)?;
-                    f.seek(SeekFrom::End(0))?;
-                    return Ok(out);
+                    if physical_keys.get(&sid).is_some_and(|existing| {
+                        existing != &(key.clone(), generation_sha256.clone())
+                    }) || logical_sids.get(&key).is_some_and(|previous| {
+                        *previous != sid && (generation_sha256.is_none() || sid < *previous)
+                    }) {
+                        return Err(std::io::Error::new(
+                            std::io::ErrorKind::InvalidData,
+                            "conflicting resolver binding",
+                        ));
+                    }
+                    physical_keys.insert(sid, (key.clone(), generation_sha256.clone()));
+                    logical_sids.insert(key, sid);
+                    let catalog_generation = generation_sha256
+                        .map(|key| {
+                            generations.get(&key).cloned().ok_or_else(|| {
+                                std::io::Error::new(
+                                    std::io::ErrorKind::InvalidData,
+                                    "missing catalog generation reference",
+                                )
+                            })
+                        })
+                        .transpose()?;
+                    records.push(ResolverRecord {
+                        sid,
+                        metric,
+                        attrs_fingerprint,
+                        agg_kind_canonical,
+                        catalog_generation,
+                    });
                 }
             }
+            safe_offset = file.stream_position()?;
         }
-        // Clean EOF — seek back to end for future appends and return.
-        f.seek(SeekFrom::End(0))?;
-        Ok(out)
+        file.seek(SeekFrom::End(0))?;
+        *self.generations.lock().unwrap() = generations;
+        Ok(records)
     }
 }
 
 /// Outcome of attempting to read a single WAL record. `Torn` means a
-/// short read or out-of-range field length was detected mid-record;
+/// short read was detected mid-record; malformed complete fields are corruption.
+/// Only an incomplete tail may be discarded;
 /// the caller truncates the file to the last `Ok` offset.
 enum ReadOne {
     Ok(ResolverRecord, u64),
     Eof,
     Torn,
+    Corrupt,
+    Io(std::io::Error),
 }
 
 fn read_one_record(f: &mut File) -> ReadOne {
@@ -478,62 +722,89 @@ fn read_one_record(f: &mut File) -> ReadOne {
     match f.read_exact(&mut sid_buf) {
         Ok(()) => {}
         Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return ReadOne::Eof,
-        Err(_) => return ReadOne::Torn,
+        Err(error) => return ReadOne::Io(error),
     }
     let sid = u64::from_le_bytes(sid_buf);
+    if sid == 0 {
+        return ReadOne::Corrupt;
+    }
 
     let mut len_buf = [0u8; 4];
-    if f.read_exact(&mut len_buf).is_err() {
-        return ReadOne::Torn;
+    if let Err(error) = f.read_exact(&mut len_buf) {
+        return if error.kind() == std::io::ErrorKind::UnexpectedEof {
+            ReadOne::Torn
+        } else {
+            ReadOne::Io(error)
+        };
     }
     let metric_len = u32::from_le_bytes(len_buf) as usize;
     if metric_len > MAX_METRIC_LEN {
-        return ReadOne::Torn;
+        return ReadOne::Corrupt;
     }
     let mut metric_bytes = vec![0u8; metric_len];
-    if f.read_exact(&mut metric_bytes).is_err() {
-        return ReadOne::Torn;
+    if let Err(error) = f.read_exact(&mut metric_bytes) {
+        return if error.kind() == std::io::ErrorKind::UnexpectedEof {
+            ReadOne::Torn
+        } else {
+            ReadOne::Io(error)
+        };
     }
 
-    if f.read_exact(&mut len_buf).is_err() {
-        return ReadOne::Torn;
+    if let Err(error) = f.read_exact(&mut len_buf) {
+        return if error.kind() == std::io::ErrorKind::UnexpectedEof {
+            ReadOne::Torn
+        } else {
+            ReadOne::Io(error)
+        };
     }
     let fp_len = u32::from_le_bytes(len_buf) as usize;
     if fp_len > MAX_FP_LEN {
-        return ReadOne::Torn;
+        return ReadOne::Corrupt;
     }
     let mut fp_bytes = vec![0u8; fp_len];
-    if f.read_exact(&mut fp_bytes).is_err() {
-        return ReadOne::Torn;
+    if let Err(error) = f.read_exact(&mut fp_bytes) {
+        return if error.kind() == std::io::ErrorKind::UnexpectedEof {
+            ReadOne::Torn
+        } else {
+            ReadOne::Io(error)
+        };
     }
 
-    if f.read_exact(&mut len_buf).is_err() {
-        return ReadOne::Torn;
+    if let Err(error) = f.read_exact(&mut len_buf) {
+        return if error.kind() == std::io::ErrorKind::UnexpectedEof {
+            ReadOne::Torn
+        } else {
+            ReadOne::Io(error)
+        };
     }
     let agg_kind_len = u32::from_le_bytes(len_buf) as usize;
     if agg_kind_len > MAX_AGG_KIND_LEN {
-        return ReadOne::Torn;
+        return ReadOne::Corrupt;
     }
     let mut agg_kind_bytes = vec![0u8; agg_kind_len];
-    if f.read_exact(&mut agg_kind_bytes).is_err() {
-        return ReadOne::Torn;
+    if let Err(error) = f.read_exact(&mut agg_kind_bytes) {
+        return if error.kind() == std::io::ErrorKind::UnexpectedEof {
+            ReadOne::Torn
+        } else {
+            ReadOne::Io(error)
+        };
     }
 
     let metric = match String::from_utf8(metric_bytes) {
         Ok(s) => s,
-        Err(_) => return ReadOne::Torn,
+        Err(_) => return ReadOne::Corrupt,
     };
     let attrs_fingerprint = match String::from_utf8(fp_bytes) {
         Ok(s) => s,
-        Err(_) => return ReadOne::Torn,
+        Err(_) => return ReadOne::Corrupt,
     };
     let agg_kind_canonical = match String::from_utf8(agg_kind_bytes) {
         Ok(s) => s,
-        Err(_) => return ReadOne::Torn,
+        Err(_) => return ReadOne::Corrupt,
     };
     let new_offset = match f.stream_position() {
         Ok(p) => p,
-        Err(_) => return ReadOne::Torn,
+        Err(error) => return ReadOne::Io(error),
     };
     ReadOne::Ok(
         ResolverRecord {
@@ -541,6 +812,7 @@ fn read_one_record(f: &mut File) -> ReadOne {
             metric,
             attrs_fingerprint,
             agg_kind_canonical,
+            catalog_generation: None,
         },
         new_offset,
     )
@@ -645,6 +917,145 @@ mod persistence_tests {
         dir.path().join("series_resolver.wal")
     }
 
+    fn generation() -> asap_types::sds::CatalogGeneration {
+        asap_types::sds::CatalogGeneration {
+            schema_version: 1,
+            plan_id: 7,
+            plan_version: 2,
+            snapshot_sha256: "new-catalog".into(),
+        }
+    }
+
+    #[test]
+    fn catalog_rotation_is_once_and_survives_restart_with_provenance() {
+        let dir = TempDir::new().unwrap();
+        let path = wal_path(&dir);
+        let resolver = Arc::new(SeriesIdResolver::open(path.clone()).unwrap());
+        let previous = resolver.resolve("m", "group=a", TEST_AGG);
+        let threads: Vec<_> = (0..8)
+            .map(|_| {
+                let resolver = Arc::clone(&resolver);
+                std::thread::spawn(move || {
+                    resolver.rotate_for_catalog_activation(
+                        "m",
+                        "group=a",
+                        TEST_AGG,
+                        previous,
+                        &generation(),
+                    )
+                })
+            })
+            .collect();
+        let ids: Vec<_> = threads
+            .into_iter()
+            .map(|thread| thread.join().unwrap())
+            .collect();
+        let successful: Vec<_> = ids
+            .iter()
+            .filter_map(|result| result.as_ref().ok())
+            .copied()
+            .collect();
+        assert_eq!(successful.len(), 1);
+        assert_ne!(successful[0], previous);
+        assert!(ids
+            .iter()
+            .filter_map(|result| result.as_ref().err())
+            .all(|error| error.kind() == std::io::ErrorKind::WouldBlock));
+        drop(resolver);
+        let reopened = SeriesIdResolver::open(path.clone()).unwrap();
+        assert_eq!(reopened.resolve("m", "group=a", TEST_AGG), successful[0]);
+        let records = FilePersistence::open(path).unwrap().replay().unwrap();
+        assert_eq!(records.len(), 2);
+        assert_eq!(
+            records[1].catalog_generation.as_deref(),
+            Some(&generation())
+        );
+    }
+
+    #[test]
+    fn failed_catalog_rotation_keeps_old_resolver_mapping() {
+        struct RejectRotation;
+        impl SeriesResolverPersistence for RejectRotation {
+            fn append(&self, _: u64, _: &str, _: &str, _: &str) -> std::io::Result<()> {
+                Ok(())
+            }
+            fn replay(&self) -> std::io::Result<Vec<ResolverRecord>> {
+                Ok(vec![])
+            }
+        }
+        let resolver = SeriesIdResolver::with_persistence(Arc::new(RejectRotation));
+        let previous = resolver.resolve("m", "group=a", TEST_AGG);
+        assert!(resolver
+            .rotate_for_catalog_activation("m", "group=a", TEST_AGG, previous, &generation())
+            .is_err());
+        assert_eq!(resolver.resolve("m", "group=a", TEST_AGG), previous);
+    }
+
+    #[test]
+    fn legacy_v3_bindings_migrate_without_changing_physical_ids() {
+        let dir = TempDir::new().unwrap();
+        let path = wal_path(&dir);
+        let mut file = File::create(&path).unwrap();
+        file.write_all(LEGACY_WAL_MAGIC).unwrap();
+        file.write_all(&9u64.to_le_bytes()).unwrap();
+        for field in ["m", "group=a", TEST_AGG] {
+            file.write_all(&(field.len() as u32).to_le_bytes()).unwrap();
+            file.write_all(field.as_bytes()).unwrap();
+        }
+        file.sync_all().unwrap();
+        drop(file);
+        let resolver = SeriesIdResolver::open(path.clone()).unwrap();
+        assert_eq!(resolver.resolve("m", "group=a", TEST_AGG), 9);
+        assert_eq!(resolver.resolve("m", "group=b", TEST_AGG), 10);
+        assert_eq!(&std::fs::read(path).unwrap()[..8], WAL_MAGIC);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn legacy_io_failure_is_not_an_incomplete_tail() {
+        let dir = TempDir::new().unwrap();
+        let mut unreadable_stream = File::open(dir.path()).unwrap();
+        assert!(matches!(
+            read_one_record(&mut unreadable_stream),
+            ReadOne::Io(_)
+        ));
+    }
+
+    #[test]
+    fn corrupt_legacy_record_preserves_original_file() {
+        let dir = TempDir::new().unwrap();
+        let path = wal_path(&dir);
+        let mut bytes = LEGACY_WAL_MAGIC.to_vec();
+        bytes.extend_from_slice(&1u64.to_le_bytes());
+        bytes.extend_from_slice(&((MAX_METRIC_LEN + 1) as u32).to_le_bytes());
+        std::fs::write(&path, &bytes).unwrap();
+        assert!(FilePersistence::open(path.clone()).is_err());
+        assert_eq!(std::fs::read(path).unwrap(), bytes);
+    }
+
+    #[test]
+    fn replay_rejects_invalid_or_conflicting_physical_bindings() {
+        for second in [0, 1] {
+            let dir = TempDir::new().unwrap();
+            let path = wal_path(&dir);
+            let persistence = FilePersistence::open(path.clone()).unwrap();
+            persistence.append(1, "first", "", TEST_AGG).unwrap();
+            let mut file = OpenOptions::new().append(true).open(&path).unwrap();
+            write_wal_record(
+                &mut file,
+                &WalRecord::Binding {
+                    sid: second,
+                    metric: "other".into(),
+                    attrs_fingerprint: "".into(),
+                    agg_kind_canonical: TEST_AGG.into(),
+                    generation_sha256: None,
+                },
+            )
+            .unwrap();
+            assert!(persistence.replay().is_err());
+        }
+    }
+
     #[test]
     fn empty_log_replays_empty() {
         let dir = TempDir::new().unwrap();
@@ -692,22 +1103,17 @@ mod persistence_tests {
     fn torn_record_truncated_on_replay() {
         let dir = TempDir::new().unwrap();
         let path = wal_path(&dir);
-        // Write two clean records, then a torn third one (sid + len
-        // header but truncated payload).
+        // Write two clean frames, then a length prefix with no payload.
         {
             let p = FilePersistence::open(path.clone()).unwrap();
             p.append(1, "m", "k=v;", TEST_AGG).unwrap();
             p.append(2, "m", "k=w;", TEST_AGG).unwrap();
         }
-        // Manually append a torn record: sid (8B) + metric_len=999
-        // (claims 999 bytes of metric but we write 0 bytes after).
+        // The third frame claims 999 bytes but has no payload.
         {
             use std::io::Write;
             let mut f = OpenOptions::new().append(true).open(&path).unwrap();
-            f.write_all(&3u64.to_le_bytes()).unwrap();
             f.write_all(&999u32.to_le_bytes()).unwrap();
-            // No payload bytes — replay reads metric_len=999 then
-            // hits EOF.
             f.sync_all().unwrap();
         }
         let pre_size = std::fs::metadata(&path).unwrap().len();
@@ -730,10 +1136,9 @@ mod persistence_tests {
     }
 
     #[test]
-    fn out_of_range_metric_len_treated_as_torn() {
-        // A corrupted file might claim a 4GB metric name. The replay
-        // must NOT allocate that much; the bounds check rejects it as
-        // torn instead.
+    fn oversized_frame_fails_without_allocating_or_discarding_history() {
+        // An oversized frame is corruption, not a torn tail. Reject it
+        // before allocation and leave the durable file unchanged.
         let dir = TempDir::new().unwrap();
         let path = wal_path(&dir);
         {
@@ -743,15 +1148,17 @@ mod persistence_tests {
         {
             use std::io::Write;
             let mut f = OpenOptions::new().append(true).open(&path).unwrap();
-            f.write_all(&2u64.to_le_bytes()).unwrap();
-            // metric_len = MAX_METRIC_LEN + 1 — over the cap.
-            f.write_all(&((MAX_METRIC_LEN as u32) + 1).to_le_bytes())
+            f.write_all(&((MAX_WAL_RECORD_BYTES as u32) + 1).to_le_bytes())
                 .unwrap();
             f.sync_all().unwrap();
         }
-        let p = FilePersistence::open(path).unwrap();
-        let records = p.replay().unwrap();
-        assert_eq!(records.len(), 1);
+        let before = std::fs::metadata(&path).unwrap().len();
+        let p = FilePersistence::open(path.clone()).unwrap();
+        assert_eq!(
+            p.replay().unwrap_err().kind(),
+            std::io::ErrorKind::InvalidData
+        );
+        assert_eq!(std::fs::metadata(path).unwrap().len(), before);
     }
 
     #[test]
@@ -830,10 +1237,17 @@ mod persistence_tests {
             }
         }
         let r = SeriesIdResolver::with_persistence(Arc::new(FailingPersistence));
+        assert!(r
+            .resolve_with_reactivation("strict", "k=v;", TEST_AGG, |_| Ok(None))
+            .is_err());
+        assert!(!r
+            .cache
+            .contains_key(&("strict".into(), "k=v;".into(), TEST_AGG.into())));
         let sid = r.resolve("m", "k=v;", TEST_AGG);
-        assert_eq!(sid, 1, "resolver returns the sid despite persistence error");
-        // Second call hits the cache; no second append attempt.
+        assert!(sid > 0, "legacy resolver returns an uncached ephemeral sid");
+        assert!(r.try_resolve("m", "k=v;", TEST_AGG).is_err());
+        assert_eq!(r.lookup("m", "k=v;", TEST_AGG), None);
         let sid2 = r.resolve("m", "k=v;", TEST_AGG);
-        assert_eq!(sid, sid2);
+        assert_ne!(sid, sid2);
     }
 }
diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs
index d94911ce9..3d64b8eeb 100644
--- a/data_plane/src/precompute_engine/maintenance_runtime.rs
+++ b/data_plane/src/precompute_engine/maintenance_runtime.rs
@@ -431,6 +431,7 @@ impl MaintenanceDagSink {
                 .map_err(schedule_error)?;
                 let mut target_output = output.clone();
                 target_output.policy_fp = target.into();
+                target_output.series_id = None;
                 derived.push((
                     Some((key, horizon_ms)),
                     target_output,
diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs
index 6bf61842e..cf88d0ea6 100644
--- a/data_plane/src/precompute_engine/output_sink.rs
+++ b/data_plane/src/precompute_engine/output_sink.rs
@@ -162,9 +162,43 @@ impl SketchStoreSink {
         let agg_cfg = &agg_cfg;
         let resolver = self.series_resolver.clone();
         let persist = || {
+            if let Some(sid) = output.series_id {
+                return self
+                    .sketch_index
+                    .ingest_precompute_with_series_id(sid, agg_cfg, output, accumulator)
+                    .inspect(|_| {
+                        crate::precompute_engine::metrics::record_materialized_outputs(1)
+                    });
+            }
+            self.sketch_index
+                .validate_routed_catalog_generation(output.catalog_generation.as_deref())
+                .ok()?;
             self.sketch_index
                 .ingest_precompute_for_agg_config(
-                    |metric, fp, ak| resolver.resolve(metric, fp, ak),
+                    |metric, fp, ak| {
+                        resolver
+                            .resolve_with_reactivation(metric, fp, ak, |sid| {
+                                self.sketch_index.validate_routed_catalog_generation(
+                                    output.catalog_generation.as_deref(),
+                                )?;
+                                let activation = self
+                                    .sketch_index
+                                    .authorize_series_reactivation(sid, output.policy_fp.into())?;
+                                if let Some(generation) = &activation {
+                                    if output.catalog_generation.as_deref()
+                                        != Some(generation.as_ref())
+                                    {
+                                        return Err(
+                                            "unbound or stale output cannot reactivate a series"
+                                                .into(),
+                                        );
+                                    }
+                                }
+                                Ok(activation)
+                            })
+                            .map_err(|error| warn!(%error, "series reactivation rejected"))
+                            .ok()
+                    },
                     agg_cfg,
                     output,
                     accumulator,
@@ -449,6 +483,100 @@ mod tests {
         );
     }
 
+    #[test]
+    fn sink_reactivates_catalog_series_without_reusing_retired_payload() {
+        let cfg = sum_agg_config(7, "cpu_seconds", &[]);
+        let fingerprint = cfg.policy_fingerprint();
+        let catalog = asap_types::summary_catalog::SummaryCatalog::from_materializations(
+            1,
+            1,
+            &[cfg.clone()],
+        )
+        .unwrap();
+        let store = Arc::new(SketchStore::new());
+        store
+            .install_summary_catalog(Arc::new(catalog.clone()))
+            .unwrap();
+        let temporary = tempfile::tempdir().unwrap();
+        let resolver =
+            Arc::new(SeriesIdResolver::open(temporary.path().join("resolver.wal")).unwrap());
+        let sink = SketchStoreSink::new(
+            store.clone(),
+            HotReloadStreamingConfig::new(StreamingConfig::new(HashMap::from([(
+                fingerprint.0,
+                cfg,
+            )]))),
+            resolver,
+        );
+        let original_generation = Arc::new(catalog.reference().unwrap());
+        let output = || {
+            let mut output = PrecomputedOutput::new(1000, 2000, None, fingerprint);
+            output.catalog_generation = Some(Arc::clone(&original_generation));
+            output
+        };
+        sink.emit_batch(vec![(output(), Box::new(SumAccumulator::with_sum(7.0)))])
+            .unwrap();
+        let old_sid = store.series_ids_for_policy(fingerprint)[0];
+        store.remove_instance(old_sid).unwrap();
+        assert!(sink
+            .emit_batch(vec![(output(), Box::new(SumAccumulator::with_sum(11.0)))])
+            .is_err());
+        let mut stale_output = output();
+        stale_output.series_id = Some(old_sid);
+        stale_output.catalog_generation = Some(Arc::new(catalog.reference().unwrap()));
+        let mut next = catalog;
+        next.plan_version += 1;
+        let next_generation = Arc::new(next.reference().unwrap());
+        store.install_summary_catalog(Arc::new(next)).unwrap();
+        assert!(sink
+            .emit_batch(vec![(
+                stale_output.clone(),
+                Box::new(SumAccumulator::with_sum(99.0))
+            )])
+            .is_err());
+        let mut next_output = output();
+        next_output.catalog_generation = Some(next_generation);
+        sink.emit_batch(vec![(
+            next_output,
+            Box::new(SumAccumulator::with_sum(11.0)),
+        )])
+        .unwrap();
+        assert!(sink
+            .emit_batch(vec![(
+                stale_output,
+                Box::new(SumAccumulator::with_sum(99.0))
+            )])
+            .is_err());
+        let new_sid = store.series_ids_for_policy(fingerprint)[0];
+        // A derived/unbound stale output must not reuse an already rotated cache hit.
+        assert!(sink
+            .emit_batch(vec![(output(), Box::new(SumAccumulator::with_sum(101.0)))])
+            .is_err());
+        let mut stale_routed_output = output();
+        stale_routed_output.series_id = Some(new_sid);
+        assert!(sink
+            .emit_batch(vec![(
+                stale_routed_output,
+                Box::new(SumAccumulator::with_sum(103.0))
+            )])
+            .is_err());
+        let mut missing_generation = PrecomputedOutput::new(1000, 2000, None, fingerprint);
+        missing_generation.series_id = Some(new_sid);
+        assert!(sink
+            .emit_batch(vec![(
+                missing_generation,
+                Box::new(SumAccumulator::with_sum(107.0))
+            )])
+            .is_err());
+        assert_ne!(old_sid, new_sid);
+        assert!(store.query_exact_agg_range(old_sid, 1000, 2000).is_empty());
+        let values = store.query_exact_agg_range(new_sid, 1000, 2000);
+        assert_eq!(
+            values[0].1.values().next().unwrap().aux_stats().sum,
+            Some(11.0)
+        );
+    }
+
     #[test]
     fn sketch_policy_is_registered_and_stored_as_sketch_state() {
         let mut cfg = sum_agg_config(8, "latency", &[]);
diff --git a/data_plane/src/precompute_engine/series_router.rs b/data_plane/src/precompute_engine/series_router.rs
index b0de9df62..bc1c319f1 100644
--- a/data_plane/src/precompute_engine/series_router.rs
+++ b/data_plane/src/precompute_engine/series_router.rs
@@ -25,10 +25,12 @@ use xxhash_rust::xxh64::xxh64;
 /// by sid without losing the data the legacy `(agg_id, group_key)` shape
 /// carried.
 pub enum WorkerMessage {
-    /// Receipt allocated after queue reservation and before any input is visible.
-    Admitted {
+    /// Immutable producer generation captured before routing. The optional
+    /// receipt proves atomic admission; absent receipts are never fabricated.
+    BoundInput {
         input: Box<WorkerMessage>,
-        revision: Arc<crate::storage_engines::types::SummaryInputRevision>,
+        generation: Arc<asap_types::sds::CatalogGeneration>,
+        revision: Option<Arc<crate::storage_engines::types::SummaryInputRevision>>,
     },
     /// A batch of samples for the same series, routed by series key.
     /// Used in `pass_raw_samples` mode where no aggregation is needed.
@@ -101,10 +103,15 @@ pub enum WorkerMessage {
 impl fmt::Debug for WorkerMessage {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            Self::Admitted { input, revision } => f
-                .debug_struct("Admitted")
+            Self::BoundInput {
+                input,
+                generation,
+                revision,
+            } => f
+                .debug_struct("BoundInput")
                 .field("input", input)
-                .field("revision", &revision.revision)
+                .field("generation", generation)
+                .field("revision", &revision.as_ref().map(|value| value.revision))
                 .finish(),
             Self::RawSamples {
                 series_key,
@@ -171,12 +178,13 @@ impl SeriesRouter {
         &self,
         messages: Vec<WorkerMessage>,
         _ingest_received_at: Instant,
+        generation: Option<Arc<asap_types::sds::CatalogGeneration>>,
     ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         // Group messages by target worker index
         let mut per_worker: HashMap<usize, Vec<WorkerMessage>> = HashMap::new();
         for msg in messages {
             let worker_idx = match &msg {
-                WorkerMessage::Admitted { .. } => {
+                WorkerMessage::BoundInput { .. } => {
                     return Err("input must be admitted by the router".into())
                 }
                 WorkerMessage::GroupSamples { sid, .. } => self.worker_for_sid(*sid),
@@ -184,7 +192,15 @@ impl SeriesRouter {
                 WorkerMessage::RawSamples { series_key, .. } => self.worker_for(series_key),
                 _ => 0,
             };
-            per_worker.entry(worker_idx).or_default().push(msg);
+            let message = match &generation {
+                Some(generation) => WorkerMessage::BoundInput {
+                    input: Box::new(msg),
+                    generation: Arc::clone(generation),
+                    revision: None,
+                },
+                None => msg,
+            };
+            per_worker.entry(worker_idx).or_default().push(message);
         }
 
         // Send to each worker concurrently
@@ -234,7 +250,7 @@ impl SeriesRouter {
                 WorkerMessage::GroupSamples { sid, .. }
                 | WorkerMessage::AccumulatorInput { sid, .. } => self.worker_for_sid(*sid),
                 WorkerMessage::RawSamples { series_key, .. } => self.worker_for(series_key),
-                WorkerMessage::Admitted { .. } => {
+                WorkerMessage::BoundInput { .. } => {
                     return Err(TryRouteError::Admission("input already admitted".into()))
                 }
                 WorkerMessage::Flush | WorkerMessage::Drain(_) | WorkerMessage::Shutdown => 0,
@@ -251,9 +267,10 @@ impl SeriesRouter {
         let revision = admit().map_err(TryRouteError::Admission)?;
         for (permit, message) in pending {
             permit.send(match &revision {
-                Some(revision) => WorkerMessage::Admitted {
+                Some(revision) => WorkerMessage::BoundInput {
                     input: Box::new(message),
-                    revision: Arc::clone(revision),
+                    generation: Arc::clone(&revision.generation),
+                    revision: Some(Arc::clone(revision)),
                 },
                 None => message,
             });
diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs
index 6fdbeed50..f58d6dd3b 100644
--- a/data_plane/src/precompute_engine/worker.rs
+++ b/data_plane/src/precompute_engine/worker.rs
@@ -37,6 +37,8 @@ use tracing::{debug, debug_span, info, warn};
 /// producing one output per (sid, window) — exactly like Arroyo's
 /// `GROUP BY window, key`.
 struct GroupState {
+    series_id: u64,
+    catalog_generation: Option<Arc<asap_types::sds::CatalogGeneration>>,
     input_revisions: BTreeMap<i64, Arc<crate::storage_engines::types::SummaryInputRevision>>,
     config: Arc<AggregationConfig>,
     /// Source policy fingerprint that minted this sid. Held so
@@ -149,6 +151,7 @@ pub struct WorkerRuntimeConfig {
 /// `SeriesIdResolver`, so one sid uniquely names one bucket.
 pub struct Worker {
     current_input_revision: Option<Arc<crate::storage_engines::types::SummaryInputRevision>>,
+    current_catalog_generation: Option<Arc<asap_types::sds::CatalogGeneration>>,
     id: usize,
     receiver: mpsc::Receiver<WorkerMessage>,
     output_sink: Arc<dyn OutputSink>,
@@ -205,6 +208,7 @@ impl Worker {
         } = runtime_config;
         Self {
             current_input_revision: None,
+            current_catalog_generation: None,
             id,
             receiver,
             output_sink,
@@ -239,17 +243,31 @@ impl Worker {
         let mut processing_error: Option<String> = None;
         while let Some(msg) = self.receiver.recv().await {
             let msg = match msg {
-                WorkerMessage::Admitted { input, revision } => {
-                    self.current_input_revision = Some(revision);
+                WorkerMessage::BoundInput {
+                    input,
+                    generation,
+                    revision,
+                } => {
+                    if revision
+                        .as_ref()
+                        .is_some_and(|receipt| receipt.generation != generation)
+                    {
+                        processing_error =
+                            Some("input receipt differs from captured generation".into());
+                        continue;
+                    }
+                    self.current_catalog_generation = Some(generation);
+                    self.current_input_revision = revision;
                     *input
                 }
                 message => {
                     self.current_input_revision = None;
+                    self.current_catalog_generation = None;
                     message
                 }
             };
             match msg {
-                WorkerMessage::Admitted { .. } => {
+                WorkerMessage::BoundInput { .. } => {
                     processing_error = Some("nested admission receipt".into());
                 }
                 WorkerMessage::GroupSamples {
@@ -405,6 +423,8 @@ impl Worker {
             let cfg = snap.get_aggregation_config(policy_fp.as_u64())?;
             let config = Arc::new(cfg.clone());
             let gs = GroupState {
+                series_id: sid,
+                catalog_generation: self.current_catalog_generation.clone(),
                 input_revisions: BTreeMap::new(),
                 window_manager: WindowManager::with_layout(
                     config.window_size,
@@ -582,6 +602,8 @@ impl Worker {
                                 PolicyFingerprint::from_config(&state.config),
                                 group_key,
                                 &state.input_revisions,
+                                state.series_id,
+                                state.catalog_generation.as_ref(),
                             );
                             emit_batch.push((output, updater.take_accumulator()));
                             debug!(
@@ -630,6 +652,8 @@ impl Worker {
                     PolicyFingerprint::from_config(&state.config),
                     group_key,
                     &state.input_revisions,
+                    state.series_id,
+                    state.catalog_generation.as_ref(),
                 );
                 emit_batch.push((output, accumulator));
             }
@@ -759,6 +783,8 @@ impl Worker {
                             PolicyFingerprint::from_config(&state.config),
                             group_key,
                             &state.input_revisions,
+                            state.series_id,
+                            state.catalog_generation.as_ref(),
                         );
                         emit_batch.push((output, incoming.clone_boxed_core()));
                     }
@@ -805,6 +831,8 @@ impl Worker {
                     PolicyFingerprint::from_config(&state.config),
                     group_key,
                     &state.input_revisions,
+                    state.series_id,
+                    state.catalog_generation.as_ref(),
                 );
                 emit_batch.push((output, accumulator));
             }
@@ -821,6 +849,8 @@ impl Worker {
                     PolicyFingerprint::from_config(&state.config),
                     group_key,
                     &state.input_revisions,
+                    state.series_id,
+                    state.catalog_generation.as_ref(),
                 );
                 emit_batch.push((output, accumulator));
             }
@@ -1011,6 +1041,8 @@ impl Worker {
                         PolicyFingerprint::from_config(&state.config),
                         &group_key,
                         &state.input_revisions,
+                        state.series_id,
+                        state.catalog_generation.as_ref(),
                     );
                     emit_batch.push((output, accumulator));
                 }
@@ -1026,6 +1058,8 @@ impl Worker {
                         PolicyFingerprint::from_config(&state.config),
                         &group_key,
                         &state.input_revisions,
+                        state.series_id,
+                        state.catalog_generation.as_ref(),
                     );
                     emit_batch.push((output, accumulator));
                 }
@@ -1113,6 +1147,8 @@ impl Worker {
                         PolicyFingerprint::from_config(&state.config),
                         &group_key,
                         &state.input_revisions,
+                        state.series_id,
+                        state.catalog_generation.as_ref(),
                     );
                     emit_batch.push((output, accumulator));
                 }
@@ -1128,6 +1164,8 @@ impl Worker {
                         PolicyFingerprint::from_config(&state.config),
                         &group_key,
                         &state.input_revisions,
+                        state.series_id,
+                        state.catalog_generation.as_ref(),
                     );
                     emit_batch.push((output, accumulator));
                 }
@@ -1248,9 +1286,13 @@ fn precomputed_output_for_group(
     policy_fp: PolicyFingerprint,
     group_key: &GroupKey,
     input_revisions: &BTreeMap<i64, Arc<crate::storage_engines::types::SummaryInputRevision>>,
+    series_id: u64,
+    catalog_generation: Option<&Arc<asap_types::sds::CatalogGeneration>>,
 ) -> PrecomputedOutput {
     let mut output = PrecomputedOutput::new(start_timestamp, end_timestamp, Some(key), policy_fp)
         .with_population_labels(population_labels_from_group_key(group_key));
+    output.series_id = Some(series_id);
+    output.catalog_generation = catalog_generation.cloned();
     output.input_revision = i64::try_from(start_timestamp)
         .ok()
         .and_then(|start| input_revisions.get(&start).cloned());
@@ -2749,6 +2791,8 @@ aggregations:
             PolicyFingerprint(7),
             &group,
             &BTreeMap::new(),
+            1,
+            None,
         );
         assert_eq!(
             output.population_labels,
diff --git a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs
index ad01a9070..a5f6c8375 100644
--- a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs
+++ b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs
@@ -124,7 +124,9 @@ fn resolve_backfill_bucket_sid(
     resolver: &SeriesIdResolver,
     config: &AggregationConfig,
     series_key: &str,
-) -> u64 {
+    store: Option<&crate::storage_engines::sketch_db::index::SketchStore>,
+    captured_generation: Option<&asap_types::sds::CatalogGeneration>,
+) -> Result<u64, String> {
     let labels = parse_labels_from_series_key(series_key);
     let grouping_pairs: Vec<(&str, &str)> = config
         .grouping_labels
@@ -135,7 +137,20 @@ fn resolve_backfill_bucket_sid(
     let attrs_fp = canonical_attrs_fingerprint(&grouping_pairs);
     let agg_kind_canonical =
         crate::storage_engines::sketch_db::data::materialization_kind_for_config(config);
-    resolver.resolve(&config.metric, &attrs_fp, &agg_kind_canonical)
+    resolver.resolve_with_reactivation(&config.metric, &attrs_fp, &agg_kind_canonical, |sid| {
+        store.map_or(Ok(None), |store| {
+            store.validate_routed_catalog_generation(captured_generation)?;
+            let activation =
+                store.authorize_series_reactivation(sid, config.policy_fingerprint().into())?;
+            if activation
+                .as_deref()
+                .is_some_and(|generation| Some(generation) != captured_generation)
+            {
+                return Err("stale backfill job cannot reactivate series".into());
+            }
+            Ok(activation)
+        })
+    })
 }
 
 /// Fallback bucket id for the resolver-less code path (registry-only
@@ -180,6 +195,7 @@ pub struct BackfillWindowProcessor {
     /// The job this processor is running on behalf of. Threaded
     /// into provenance calls; never used for dispatch logic.
     job_id: u64,
+    catalog_generation: Option<Arc<asap_types::sds::CatalogGeneration>>,
 }
 
 impl BackfillWindowProcessor {
@@ -194,6 +210,7 @@ impl BackfillWindowProcessor {
             series_resolver: None,
             registry,
             job_id,
+            catalog_generation: None,
         }
     }
 
@@ -203,6 +220,7 @@ impl BackfillWindowProcessor {
         mut self,
         sketch_index: Arc<crate::storage_engines::sketch_db::index::SketchStore>,
     ) -> Self {
+        self.catalog_generation = sketch_index.active_catalog_generation();
         self.sketch_index = Some(sketch_index);
         self
     }
@@ -279,7 +297,13 @@ impl WindowProcessor for BackfillWindowProcessor {
         for sample in samples {
             let group_key = extract_group_key(&sample.labels, &config);
             let sid = match resolver_opt {
-                Some(r) => resolve_backfill_bucket_sid(r.as_ref(), &config, &sample.labels),
+                Some(r) => resolve_backfill_bucket_sid(
+                    r.as_ref(),
+                    &config,
+                    &sample.labels,
+                    self.sketch_index.as_deref(),
+                    self.catalog_generation.as_deref(),
+                )?,
                 None => fallback_bucket_id(&group_key),
             };
             by_bucket
@@ -328,13 +352,15 @@ impl WindowProcessor for BackfillWindowProcessor {
             } else {
                 Some(build_group_key_label_values(&group_key))
             };
-            let output = crate::storage_engines::types::PrecomputedOutput::new_backfilled(
+            let mut output = crate::storage_engines::types::PrecomputedOutput::new_backfilled(
                 window_range.0,
                 window_range.1,
                 key,
                 self.job_id,
                 PolicyFingerprint::from_config(&config),
             );
+            output.series_id = Some(sid);
+            output.catalog_generation = self.catalog_generation.clone();
             batch.push((sid, output, accumulator));
         }
 
@@ -361,7 +387,8 @@ impl WindowProcessor for BackfillWindowProcessor {
                             &config,
                             output,
                             accumulator.as_ref(),
-                        );
+                        )
+                        .ok_or("backfill summary state publication rejected")?;
                     }
                 }
                 None => {
@@ -842,6 +869,15 @@ mod tests {
         let hot = HotReloadStreamingConfig::from_arc(streaming.clone());
         let registry = Arc::new(BackfillRegistry::new());
         let sketch_index = Arc::new(SketchStore::new());
+        let catalog = asap_types::summary_catalog::SummaryCatalog::from_materializations(
+            1,
+            1,
+            &[cfg.clone()],
+        )
+        .unwrap();
+        sketch_index
+            .install_summary_catalog(Arc::new(catalog))
+            .unwrap();
         let resolver = Arc::new(SeriesIdResolver::new());
 
         let job_id = registry.create(
@@ -896,8 +932,10 @@ mod tests {
         // `SeriesIdResolver::lookup` would return for the same
         // `(metric, grouping-values, agg_kind)` tuple — i.e. live
         // ingest and backfill share one sid namespace.
-        let sid_a = resolve_backfill_bucket_sid(&resolver, &cfg, "latency{svc=\"a\"}");
-        let sid_b = resolve_backfill_bucket_sid(&resolver, &cfg, "latency{svc=\"b\"}");
+        let sid_a =
+            resolve_backfill_bucket_sid(&resolver, &cfg, "latency{svc=\"a\"}", None, None).unwrap();
+        let sid_b =
+            resolve_backfill_bucket_sid(&resolver, &cfg, "latency{svc=\"b\"}", None, None).unwrap();
         assert_ne!(sid_a, sid_b, "distinct svc values mint distinct sids");
         assert_eq!(sketch_index.classify(sid_a), SeriesLookup::Hit);
         assert_eq!(sketch_index.classify(sid_b), SeriesLookup::Hit);
@@ -917,8 +955,14 @@ mod tests {
         let resolver = SeriesIdResolver::new();
 
         // Backfill side: derive sid via the new helper.
-        let backfill_sid =
-            resolve_backfill_bucket_sid(&resolver, &cfg, "latency{svc=\"a\",zone=\"z0\"}");
+        let backfill_sid = resolve_backfill_bucket_sid(
+            &resolver,
+            &cfg,
+            "latency{svc=\"a\",zone=\"z0\"}",
+            None,
+            None,
+        )
+        .unwrap();
 
         // Exercise the actual live sink instead of duplicating its SID formula.
         let store = crate::storage_engines::sketch_db::index::SketchStore::new();
diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs
index 6ac76aef4..72d172468 100644
--- a/data_plane/src/storage_engines/sketch_db/index/mod.rs
+++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs
@@ -690,7 +690,15 @@ pub struct SketchStore {
     /// flush-then-evict loop is the memory bound).
     persistence_read: RwLock<Option<Arc<PersistenceReadHandle>>>,
     persistence_metadata: RwLock<Option<Arc<persistence::metadata::SidMetadataStore>>>,
-    removed_sids: RwLock<BTreeSet<u64>>,
+    removed_sids: RwLock<
+        BTreeMap<
+            u64,
+            (
+                Option<Arc<asap_types::sds::CatalogGeneration>>,
+                Option<SummaryDefinitionId>,
+            ),
+        >,
+    >,
     /// Seal cadence in distinct windows, applied to every per-sid
     /// `SidStoreData` once persistence is enabled. `0` (the default)
     /// disables cadence sealing. Set by [`Self::enable_persistence_mode`].
@@ -783,7 +791,7 @@ impl SketchStore {
             .map(str::to_owned);
         // Fixed lock order: instances → policy_to_series_ids → metric_to_series_ids.
         let mut instances = self.instances.write().unwrap();
-        if self.removed_sids.read().unwrap().contains(&sid) {
+        if self.removed_sids.read().unwrap().contains_key(&sid) {
             tracing::warn!(sid, "rejecting reuse of a removed summary instance ID");
             return;
         }
@@ -2405,16 +2413,8 @@ impl SketchStore {
                 m.first_seen_unix_ms,
             );
         if !m.policy_fp.is_unset() {
-            let (catalog, generation) = self.descriptors.authoritative_snapshot()?;
-            let definition = SummaryDefinitionId::from(m.policy_fp);
-            let identity = catalog.materializations.get(&definition)?;
-            if identity.summary_descriptor_id != *m.summary_descriptor.id()
-                || identity.data_descriptor_id != *m.data_descriptor.id()
-            {
-                return None;
-            }
-            record.summary_definition_id = Some(definition);
-            record.catalog_generation = Some(generation);
+            record.summary_definition_id = Some(SummaryDefinitionId::from(m.policy_fp));
+            record.catalog_generation = Some(Arc::clone(m.catalog_generation.as_ref()?));
         }
         record.retired_at_ms = m.retired_at_ms;
         record.expires_at_ms = m.expires_at_ms;
@@ -2487,6 +2487,62 @@ impl SketchStore {
         Some(Arc::clone(&instance.metadata))
     }
 
+    pub(crate) fn active_catalog_generation(
+        &self,
+    ) -> Option<Arc<asap_types::sds::CatalogGeneration>> {
+        self.descriptors
+            .authoritative_snapshot()
+            .map(|(_, generation)| generation)
+    }
+
+    /// Resolving a logical key may already return a replacement physical SID.
+    /// Validate producer provenance before either a cache hit or a rotation.
+    pub(crate) fn validate_routed_catalog_generation(
+        &self,
+        captured: Option<&asap_types::sds::CatalogGeneration>,
+    ) -> Result<(), String> {
+        if self.active_catalog_generation().as_deref() != captured {
+            return Err(
+                "unbound or stale producer cannot resolve the active catalog's physical series"
+                    .into(),
+            );
+        }
+        Ok(())
+    }
+
+    /// Return the current generation only when it explicitly reintroduces a
+    /// previously removed logical materialization. Ordinary same-generation
+    /// writes and missing provenance cannot start a new physical lifetime.
+    pub(crate) fn authorize_series_reactivation(
+        &self,
+        sid: u64,
+        definition: SummaryDefinitionId,
+    ) -> Result<Option<Arc<asap_types::sds::CatalogGeneration>>, String> {
+        let removed = self
+            .removed_sids
+            .read()
+            .map_err(|_| "series tombstone lock poisoned")?;
+        let Some((old_generation, old_definition)) = removed.get(&sid) else {
+            return Ok(None);
+        };
+        let (catalog, generation) = self
+            .descriptors
+            .authoritative_snapshot()
+            .ok_or("series reactivation requires an authoritative catalog")?;
+        if *old_definition != Some(definition)
+            || !catalog.materializations.contains_key(&definition)
+        {
+            return Err("series reactivation does not match the installed materialization".into());
+        }
+        let old_generation = old_generation
+            .as_ref()
+            .ok_or("removed series has no catalog provenance")?;
+        if old_generation.as_ref() == generation.as_ref() {
+            return Err("same-generation append cannot reactivate a removed series".into());
+        }
+        Ok(Some(generation))
+    }
+
     /// Drop a sid's metadata + its series state + both secondary-index
     /// entries (`policy_to_series_ids` and `metric_to_series_ids`). Mirrors
     /// `SchemaRegistry::remove_schema` for the eviction path's
@@ -2509,7 +2565,16 @@ impl SketchStore {
             let mut instances = self.instances.write().ok()?;
             if let Some(instance) = instances.get(&sid) {
                 self.persist_lifecycle(instance, true).ok()?;
-                self.removed_sids.write().ok()?.insert(sid);
+                let record = self.metadata_record(instance);
+                self.removed_sids.write().ok()?.insert(
+                    sid,
+                    (
+                        record
+                            .as_ref()
+                            .and_then(|value| value.catalog_generation.clone()),
+                        record.and_then(|value| value.summary_definition_id),
+                    ),
+                );
             }
             let mut policy_idx = self.policy_to_series_ids.write().unwrap();
             let mut metric_idx = self.metric_to_series_ids.write().unwrap();
@@ -2598,9 +2663,9 @@ impl SketchStore {
     /// once a sid is retired by [`crate::storage_engines::sketch_db::lifecycle::reconcile_from_streaming_config`]
     /// further writes are rejected here so the eviction sweep can
     /// drop residual state cleanly.
-    pub fn ingest_precompute_for_agg_config(
+    pub fn ingest_precompute_for_agg_config<R: Into<Option<u64>>>(
         &self,
-        mint_sid: impl FnOnce(&str, &str, &str) -> u64,
+        mint_sid: impl FnOnce(&str, &str, &str) -> R,
         agg_cfg: &asap_types::aggregation_config::AggregationConfig,
         output: &crate::storage_engines::types::PrecomputedOutput,
         accumulator: &dyn crate::storage_engines::types::AggregateCore,
@@ -2614,7 +2679,7 @@ impl SketchStore {
         let (attrs_fp, _label_values_map) = build_attrs_fp_and_label_map(agg_cfg, output);
         let agg_kind_canonical =
             crate::storage_engines::sketch_db::data::materialization_kind_for_config(agg_cfg);
-        let sid = mint_sid(&agg_cfg.metric, &attrs_fp, &agg_kind_canonical);
+        let sid = mint_sid(&agg_cfg.metric, &attrs_fp, &agg_kind_canonical).into()?;
         self.ingest_precompute_with_series_id(sid, agg_cfg, output, accumulator)
     }
 
@@ -2646,6 +2711,12 @@ impl SketchStore {
 
         match self.instance(sid) {
             None => {
+                if self.active_catalog_generation().as_deref()
+                    != output.catalog_generation.as_deref()
+                {
+                    return None;
+                }
+
                 let group_by_keys: BTreeSet<String> = key_names.iter().cloned().collect();
                 // PR 6 follow-up: ExactAgg-backed sids carry an
                 // `ExactAgg(agg_type)` capability so the analyzer can
@@ -2673,12 +2744,34 @@ impl SketchStore {
                     policy_fp: output.policy_fp,
                 });
             }
-            Some(existing) if !existing.is_writable() => {
+            Some(existing) if !existing.is_writable() || existing.policy_fp != output.policy_fp => {
                 return None;
             }
             Some(_) => {}
         }
 
+        // Keep the physical lifetime alive through publication. Removal takes
+        // this same lock exclusively, so it cannot race metadata validation and
+        // recreate orphan payload after the tombstone commits.
+        let instances = self.instances.read().ok()?;
+        let binding = instances.get(&sid)?;
+        if !binding.metadata.is_writable() || binding.metadata.policy_fp != output.policy_fp {
+            return None;
+        }
+        if binding.catalog_generation.is_some() && output.catalog_generation.is_none() {
+            return None;
+        }
+        if let Some(captured) = output.catalog_generation.as_deref() {
+            // Existing unchanged series may drain their birth generation or
+            // accept the currently installed generation. A replacement born in
+            // a newer generation cannot accept an older unbound cache hit.
+            if binding.catalog_generation.as_deref() != Some(captured)
+                && self.active_catalog_generation().as_deref() != Some(captured)
+            {
+                return None;
+            }
+        }
+
         if let Some(retained_windows) = agg_cfg.num_aggregates_to_retain {
             let required_horizon_ms = retained_windows
                 .saturating_mul(agg_cfg.slide_interval)
@@ -2821,7 +2914,12 @@ impl SketchStore {
                     .load()?
                     .into_iter()
                     .filter(|record| record.removed)
-                    .map(|record| record.sid),
+                    .map(|record| {
+                        (
+                            record.sid,
+                            (record.catalog_generation, record.summary_definition_id),
+                        )
+                    }),
             );
         }
         let (_loaded_manifest, report) = recovery::recover(&cfg.disk_path)?;
@@ -4550,6 +4648,92 @@ mod tests {
         assert!(store.series_ids_for_policy(fingerprint).is_empty());
     }
 
+    #[test]
+    fn catalog_reactivation_uses_new_physical_series_without_old_disk_payload() {
+        use crate::drivers::ingest::series_resolver::SeriesIdResolver;
+        let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot =
+            serde_json::from_str(include_str!(
+                "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json"
+            ))
+            .unwrap();
+        let plan = snapshot.compile().unwrap();
+        let fingerprint = plan.precompute_plan.materializations[0].policy_fingerprint();
+        let definition = fingerprint.into();
+        let mut next_catalog = plan.summary_catalog.clone();
+        next_catalog.plan_version += 1;
+        let directory = tempfile::tempdir().unwrap();
+        let disk = directory.path().join("state");
+        let wal = directory.path().join("resolver.wal");
+        let old_sid;
+        let new_sid;
+        {
+            let resolver = SeriesIdResolver::open(wal.clone()).unwrap();
+            let store = Arc::new(SketchStore::new());
+            store
+                .install_summary_catalog(Arc::new(plan.summary_catalog))
+                .unwrap();
+            let mut persistence = store.start_persistence(durable_cfg(disk.clone())).unwrap();
+            old_sid = resolver.resolve("metric", "group", "family");
+            store.register(meta_with_policy(old_sid, fingerprint));
+            for pane in 0..4 {
+                store.append_sample(
+                    old_sid,
+                    BTreeMap::new(),
+                    (pane * 30_000, (pane + 1) * 30_000),
+                    sample(1),
+                );
+            }
+            assert!(wait_until(
+                || !persistence.manifest.live_parts().is_empty(),
+                Duration::from_secs(5)
+            ));
+            let old_parts = persistence.manifest.live_parts().len();
+            store.remove_instance(old_sid).unwrap();
+            assert!(resolver
+                .resolve_with_reactivation("metric", "group", "family", |sid| store
+                    .authorize_series_reactivation(sid, definition))
+                .is_err());
+            store
+                .install_summary_catalog(Arc::new(next_catalog.clone()))
+                .unwrap();
+            new_sid = resolver
+                .resolve_with_reactivation("metric", "group", "family", |sid| {
+                    store.authorize_series_reactivation(sid, definition)
+                })
+                .unwrap();
+            assert_ne!(new_sid, old_sid);
+            store.register(meta_with_policy(new_sid, fingerprint));
+            for pane in 0..4 {
+                store.append_sample(
+                    new_sid,
+                    BTreeMap::new(),
+                    (pane * 30_000, (pane + 1) * 30_000),
+                    sample(2),
+                );
+            }
+            assert!(wait_until(
+                || persistence.manifest.live_parts().len() > old_parts,
+                Duration::from_secs(5)
+            ));
+            persistence.shutdown();
+        }
+        let resolver = SeriesIdResolver::open(wal).unwrap();
+        assert_eq!(resolver.resolve("metric", "group", "family"), new_sid);
+        let recovered = Arc::new(SketchStore::new());
+        recovered
+            .install_summary_catalog(Arc::new(next_catalog))
+            .unwrap();
+        let _persistence = recovered.start_persistence(durable_cfg(disk)).unwrap();
+        assert!(recovered.query_range(old_sid, 0, 90_000).is_empty());
+        let rows = recovered.query_range(new_sid, 0, 90_000);
+        assert!(!rows.is_empty());
+        assert!(rows
+            .iter()
+            .flat_map(|row| row.samples.values())
+            .flatten()
+            .all(|sample| sample.bytes == vec![2]));
+    }
+
     #[test]
     fn force_expire_never_extends_existing_lifecycle_deadlines() {
         let store = SketchStore::new();
diff --git a/data_plane/src/storage_engines/sketch_db/sds.rs b/data_plane/src/storage_engines/sketch_db/sds.rs
index beecf8f14..38a3d98de 100644
--- a/data_plane/src/storage_engines/sketch_db/sds.rs
+++ b/data_plane/src/storage_engines/sketch_db/sds.rs
@@ -84,6 +84,9 @@ pub struct SdsBinding {
     pub metadata: Arc<SketchInstanceMetadata>,
     pub summary_descriptor: Arc<SummaryDescriptor>,
     pub data_descriptor: Arc<DataDescriptor>,
+    /// Immutable provenance of this physical series lifetime, shared with
+    /// the catalog snapshot used when the descriptors were bound.
+    pub catalog_generation: Option<Arc<asap_types::sds::CatalogGeneration>>,
 }
 
 impl std::ops::Deref for SdsBinding {
@@ -146,8 +149,8 @@ impl SummaryDescriptorRegistry {
     }
 
     pub fn bind(&self, metadata: SketchInstanceMetadata) -> Result<SdsBinding, String> {
-        let authoritative = self.authoritative_catalog();
-        let configured = if let Some(catalog) = authoritative.as_ref() {
+        let authoritative = self.authoritative_snapshot();
+        let configured = if let Some((catalog, _)) = authoritative.as_ref() {
             if metadata.policy_fp.is_unset() {
                 return Err(
                     "materialization identity is required by the installed SummaryCatalog".into(),
@@ -209,6 +212,7 @@ impl SummaryDescriptorRegistry {
             metadata: Arc::new(metadata),
             summary_descriptor,
             data_descriptor,
+            catalog_generation: authoritative.map(|(_, generation)| generation),
         })
     }
 
diff --git a/data_plane/src/storage_engines/types/precomputed_output.rs b/data_plane/src/storage_engines/types/precomputed_output.rs
index 5a84cc215..26bcc733e 100644
--- a/data_plane/src/storage_engines/types/precomputed_output.rs
+++ b/data_plane/src/storage_engines/types/precomputed_output.rs
@@ -52,6 +52,12 @@ pub struct SummaryInputRevision {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct PrecomputedOutput {
+    /// Physical lifetime chosen before execution; never re-resolve a queued
+    /// fragment against a newer logical-series mapping.
+    #[serde(skip)]
+    pub series_id: Option<u64>,
+    #[serde(skip)]
+    pub catalog_generation: Option<std::sync::Arc<asap_types::sds::CatalogGeneration>>,
     #[serde(skip)]
     pub input_revision: Option<std::sync::Arc<SummaryInputRevision>>,
     pub start_timestamp: u64,
@@ -91,6 +97,8 @@ impl PrecomputedOutput {
         policy_fp: PolicyFingerprint,
     ) -> Self {
         Self {
+            series_id: None,
+            catalog_generation: None,
             input_revision: None,
             start_timestamp,
             end_timestamp,
@@ -118,6 +126,8 @@ impl PrecomputedOutput {
         policy_fp: PolicyFingerprint,
     ) -> Self {
         Self {
+            series_id: None,
+            catalog_generation: None,
             input_revision: None,
             start_timestamp,
             end_timestamp,
diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md
index 82fd1c4a8..abc38b97b 100644
--- a/docs/design_docs/summary-catalog-sds-architecture.md
+++ b/docs/design_docs/summary-catalog-sds-architecture.md
@@ -37,9 +37,13 @@ operator-specific stores beside `SketchStore`.
 
 The target model has descriptor registries plus pane instances. Descriptor IDs
 are derived from canonical semantic content; display names and runtime SIDs are
-not descriptor identities. The current implementation uses the canonical string
-itself as the ID. A future hashed representation must preserve the same content
-identity and handle collisions explicitly.
+not descriptor identities. `SummaryDescriptorId` and `DataDescriptorId` currently
+contain versioned canonical semantic strings. `SummaryDefinitionId` is a distinct
+typed policy fingerprint, and `CatalogGeneration` identifies a publication using
+its digest and plan version. A physical `SeriesId` identifies one storage lifetime
+of a definition/group; it is neither a descriptor ID nor a pane instance ID.
+Changing descriptor encoding to a hash must preserve content identity and handle
+collisions explicitly.
 
 ```rust
 struct SummaryDescriptor {
@@ -371,3 +375,31 @@ creates a new Data Descriptor. Advancing the time range creates a new Summary
 Instance. Merge compatibility additionally requires the operator's merge rules,
 compatible data scopes and valid instance coverage; sharing descriptors alone
 does not authorize merging overlapping observations.
+
+
+### Retired physical series and catalog reactivation
+
+A persisted removal tombstone prevents late fragments and stale metadata flushes
+from reopening the same physical `SeriesId`. A later installed catalog generation
+may authorize a fresh physical series for the same logical definition/group.
+The resolver writes that rotation and its catalog provenance before changing its
+cache; ordinary writes from the original generation cannot authorize rotation.
+The original physical ID remains tombstoned so old disk parts cannot enter the
+replacement's readout.
+
+Queued precompute inputs carry their captured catalog generation and physical
+series ID separately from an optional admission receipt. Workers preserve both
+on publication. A delayed output writes its original physical series, never a
+newly resolved replacement. Derived materializations resolve their own target
+series while retaining the source generation proof. Backfill processors capture
+the catalog generation when attached to the store; old jobs cannot authorize a
+new catalog's rotation. An older queued input that has not yet published its
+first storage instance is conservatively rejected after a catalog change. Already
+registered retained series can drain their birth generation or accept the current
+generation. Seamless re-planning of unpublished old inputs requires additional
+first-mint provenance; it is not guaranteed by this transition.
+
+This is an explicit lifetime transition, not cross-generation recovery of arbitrary
+summary state. Legacy records without trustworthy catalog provenance remain
+unbound. Tombstone reclamation still requires coordinated removal of old physical
+parts and is not implemented by this transition.