diff --git a/AGENTS.md b/AGENTS.md
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -124,6 +124,7 @@
 - `counts`: Maps `k|{NAME}` or `r|{DID}|{COL}` -> `Count` (u64 BE Bytes).
 - `filter`: Stores filter config. Handled by the `db::filter` and `db::pds_meta` modules. Includes:
     - Mode key `m` -> `FilterMode` (MessagePack).
+    - Storage mode marker `storage_mode` -> `StorageModeMarker` (MessagePack): the immutable `only_index_links`/`ephemeral` layout the database was adopted under, written on first open and enforced by `db::storage_mode`; a config flip against an existing database is rejected at startup (markerless legacy databases pass one inference check before adoption).
     - Set entries for signals (`s|{NSID}`), collections (`c|{NSID}`), and excludes (`x|{DID}`) -> empty value.
     - PDS rate tiers: `{host}|tier` -> tier name (UTF-8 string).
     - PDS host statuses: `{host}|status` -> `HostStatus` (MessagePack).
diff --git a/docs/configuration.md b/docs/configuration.md
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -19,9 +19,9 @@
 | variable | default | description |
 | :--- | :--- | :--- |
 | `FULL_NETWORK` | `false` (indexer), `true` (relay) | if `true`, discover and index all repos in the network |
-| `EPHEMERAL` | `false` (indexer), `true` (relay) | if enabled, no records are stored (in indexer mode). events are deleted after a certain duration (`EPHEMERAL_TTL`) |
+| `EPHEMERAL` | `false` (indexer), `true` (relay) | if enabled, no records are stored (in indexer mode). events are deleted after a certain duration (`EPHEMERAL_TTL`). immutable per database: the setting is persisted on first open and a later flip is rejected at startup — keep the previous setting or use a fresh database path |
 | `EPHEMERAL_TTL` | `60min`, `3d` (relay) | how long to keep events before deletion. when built with `jetstream`, retained Jetstream replay metadata is pruned on the same schedule |
-| `ONLY_INDEX_LINKS` | `false` | indexer only. if enabled, record blocks are not stored, only the index (records, counts, events) is kept. `getRecord`, `listRecords`, and `getRepo` will return errors. the event stream and Jetstream stream still work, but create/update events will not include record values |
+| `ONLY_INDEX_LINKS` | `false` | indexer only. if enabled, record blocks are not stored, only the index (records, counts, events) is kept. `getRecord`, `listRecords`, and `getRepo` will return errors. the event stream and Jetstream stream still work, but create/update events will not include record values. immutable per database: the setting is persisted on first open and a later flip is rejected at startup — keep the previous setting or use a fresh database path |
 
 ## filter
 
diff --git a/src/db/mod.rs b/src/db/mod.rs
--- a/src/db/mod.rs
+++ b/src/db/mod.rs
@@ -23,6 +23,8 @@
 mod lifecycle_counts;
 pub mod migration;
 pub mod pds_meta;
+#[cfg(feature = "indexer")]
+pub(crate) mod storage_mode;
 pub mod types;
 
 pub mod keyspaces;
diff --git a/src/db/open.rs b/src/db/open.rs
--- a/src/db/open.rs
+++ b/src/db/open.rs
@@ -43,10 +43,21 @@
 
         let this = Self::assemble_keyspaces_and_verify(&cx, count_delta_gc_watermark)?;
 
+        // a marked database can reject a mode mismatch before migrations or
+        // compaction filters mutate anything. pre-marker databases are
+        // adopted after migration, when their record layout is inferable.
+        #[cfg(feature = "indexer")]
+        let storage_mode_marked = super::storage_mode::preflight(&this, cfg)?;
+
         migration::run(&this)?;
 
-        // history compaction stays fail-safe (keep everything) until
-        // migrations have succeeded.
+        #[cfg(feature = "indexer")]
+        if !storage_mode_marked {
+            super::storage_mode::adopt(&this, cfg)?;
+        }
+
+        // history compaction stays fail-safe (keep everything) until storage
+        // mode validation and migrations have both succeeded.
         history_ttl
             .set(cfg.history_ttl)
             .map_err(|_| miette::miette!("history retention was initialized twice"))?;
diff --git a/src/db/storage_mode.rs b/src/db/storage_mode.rs
new file mode 100644
--- /dev/null
+++ b/src/db/storage_mode.rs
@@ -0,0 +1,434 @@
+//! storage-mode validation at open time.
+//!
+//! the storage layout has two immutable axes: `only_index_links` (record
+//! values are bodies vs cids) and `ephemeral` (bodies live only inside the
+//! TTL-bounded event log). neither is safely convertible in place:
+//!
+//! - flipping links-only rewrites what `records` values mean, so replay of
+//!   events written before the flip would parse cids as bodies or vice
+//!   versa. opening with a different value than the database was written
+//!   with is rejected with an actionable error.
+//! - ephemeral databases have no record heads or history at all. interpreting
+//!   one layout as the other would either discard permanent heads or retain
+//!   body bytes past the promised event TTL.
+//!
+//! mismatches are detected via a marker in the `filter` keyspace written on
+//! first open. databases that predate the marker get it written for the
+//! currently configured layout: a flip that already happened silently before
+//! the marker existed stays silent, and future flips are protected.
+
+#[cfg(feature = "indexer_stream")]
+use miette::Context;
+use miette::{IntoDiagnostic, Result};
+use serde::{Deserialize, Serialize};
+
+use crate::config::Config;
+use crate::db::Db;
+
+/// persisted marker of the storage layout the db was written with, kept in
+/// the filter keyspace next to the other instance-level config
+pub(super) const STORAGE_MODE_KEY: &[u8] = b"storage_mode";
+
+#[derive(Serialize, Deserialize, PartialEq, Eq, Debug)]
+struct StorageModeMarker {
+    links_only: bool,
+    ephemeral: bool,
+}
+
+/// validate an existing marker before migrations can write anything.
+#[cfg(feature = "indexer")]
+pub(crate) fn preflight(db: &Db, cfg: &Config) -> Result<bool> {
+    let marker = StorageModeMarker {
+        links_only: cfg.only_index_links,
+        ephemeral: cfg.ephemeral,
+    };
+
+    let Some(stored) = db.filter.get(STORAGE_MODE_KEY).into_diagnostic()? else {
+        if let Some(inferred_ephemeral) = infer_existing_ephemeral(db)?
+            && inferred_ephemeral != marker.ephemeral
+        {
+            return Err(miette::miette!(
+                "legacy event bodies imply ephemeral={inferred_ephemeral} but config requests {}: \
+                 refusing to migrate a markerless database under the wrong storage mode. keep the \
+                 previous HYDRANT_EPHEMERAL setting or use an explicit offline migration",
+                marker.ephemeral,
+            ));
+        }
+        return Ok(false);
+    };
+    let prev: StorageModeMarker = rmp_serde::from_slice(&stored).into_diagnostic()?;
+    if prev.ephemeral
+        && db
+            .indexer
+            .records
+            .iter()
+            .next()
+            .map(|guard| guard.into_inner())
+            .transpose()
+            .into_diagnostic()?
+            .is_some()
+    {
+        miette::bail!(
+            "ephemeral storage marker conflicts with persisted record heads; rebuild this unsupported pre-release database from v9"
+        );
+    }
+    if prev.links_only != marker.links_only {
+        return Err(miette::miette!(
+            "database was written with only_index_links={} but config requests {}: \
+             links-only and body storage are not convertible in place. keep the previous \
+             setting or start from a fresh database path",
+            prev.links_only,
+            marker.links_only,
+        ));
+    }
+    if prev.ephemeral != marker.ephemeral {
+        return Err(miette::miette!(
+            "database was written with ephemeral={} but config requests {}: \
+             permanent and ephemeral storage are not convertible at startup. keep the \
+             previous HYDRANT_EPHEMERAL setting, use an explicit offline migration, or \
+             start from a fresh database path",
+            prev.ephemeral,
+            marker.ephemeral,
+        ));
+    }
+    Ok(true)
+}
+
+/// infer shipped v9 mode from body-bearing stream events. ephemeral writes
+/// stored `Block`; permanent writes stored `Ptr`. body-less delete events do
+/// not identify a mode, and seeing both encodings means the database was
+/// already switched unsafely before markers existed.
+fn infer_existing_ephemeral(db: &Db) -> Result<Option<bool>> {
+    let inferred = db
+        .indexer
+        .records
+        .iter()
+        .next()
+        .map(|guard| guard.into_inner())
+        .transpose()
+        .into_diagnostic()?
+        .map(|_| false);
+
+    #[cfg(feature = "indexer_stream")]
+    {
+        let mut inferred = inferred;
+        for guard in db.stream.events.iter() {
+            let value = guard.value().into_diagnostic()?;
+            let event: crate::types::StoredEvent<'_> = rmp_serde::from_slice(&value)
+                .into_diagnostic()
+                .wrap_err("cannot infer storage mode from a legacy event")?;
+            let ephemeral = match event.data {
+                crate::types::StoredData::Block(_) => true,
+                crate::types::StoredData::Ptr(_) => false,
+                crate::types::StoredData::Nothing => continue,
+            };
+            if inferred.is_some_and(|previous| previous != ephemeral) {
+                miette::bail!(
+                    "database contains both inline ephemeral and pointer-based permanent events; use an explicit offline migration"
+                );
+            }
+            inferred = Some(ephemeral);
+        }
+        Ok(inferred)
+    }
+    #[cfg(not(feature = "indexer_stream"))]
+    {
+        Ok(inferred)
+    }
+}
+
+/// adopt the configured layout for a database that predates the marker. this
+/// runs after migrations because legacy record values alone cannot distinguish
+/// a body-bearing database from links-only storage.
+#[cfg(feature = "indexer")]
+pub(crate) fn adopt(db: &Db, cfg: &Config) -> Result<()> {
+    let marker = StorageModeMarker {
+        links_only: cfg.only_index_links,
+        ephemeral: cfg.ephemeral,
+    };
+    if let Some(inferred_links_only) = infer_existing_links_only(db)?
+        && inferred_links_only != marker.links_only
+    {
+        return Err(miette::miette!(
+            "database record values imply only_index_links={inferred_links_only} but config requests {}: \
+             refusing to bless an incompatible pre-marker database. keep the existing setting or \
+             migrate into a fresh database path",
+            marker.links_only,
+        ));
+    }
+    db.filter
+        .insert(
+            STORAGE_MODE_KEY,
+            rmp_serde::to_vec_named(&marker).into_diagnostic()?,
+        )
+        .into_diagnostic()?;
+    tracing::info!(
+        links_only = marker.links_only,
+        ephemeral = marker.ephemeral,
+        "storage mode marker initialized"
+    );
+    Ok(())
+}
+
+/// infer the layout of a database created before `STORAGE_MODE_KEY`. scanning
+/// every head is intentional: accepting a mixed keyspace would make whichever
+/// value happens to sort first define the interpretation of all other values.
+fn infer_existing_links_only(db: &Db) -> Result<Option<bool>> {
+    let mut inferred = None;
+    for guard in db.indexer.records.iter() {
+        let value = guard.value().into_diagnostic()?;
+        let links_only = crate::db::is_cid_record_value(&value);
+        if inferred.is_some_and(|previous| previous != links_only) {
+            miette::bail!(
+                "database contains mixed record body and cid-only values; refusing to infer only_index_links"
+            );
+        }
+        inferred = Some(links_only);
+    }
+    Ok(inferred)
+}
+
+#[cfg(all(test, feature = "indexer"))]
+mod tests {
+    use super::*;
+    use crate::db::types::{DbAction, DbRkey, DbTid};
+    use crate::state::AppState;
+    use jacquard_common::types::string::{Did, Tid};
+
+    fn did() -> Did<'static> {
+        Did::new("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap()
+    }
+
+    fn config_at(path: &std::path::Path, ephemeral: bool, links_only: bool) -> Config {
+        let mut config = Config::default();
+        config.database_path = path.to_path_buf();
+        config.ephemeral = ephemeral;
+        config.only_index_links = links_only;
+        config
+    }
+
+    fn write_record(state: &AppState, rkey: &str, body: &[u8]) {
+        let mut txn = crate::db::Txn::new(&state.db);
+        let did = did();
+        let rev = DbTid::from(&Tid::now_0());
+        let mut record_txn = txn.records(state, &rev, &did).unwrap();
+        let block = bytes::Bytes::copy_from_slice(body);
+        let cid = jacquard_repo::mst::util::compute_cid(block.as_ref()).unwrap();
+        record_txn
+            .put_record(
+                "app.bsky.feed.post",
+                &DbRkey::new(rkey),
+                &cid,
+                &block,
+                DbAction::Create,
+            )
+            .unwrap();
+        record_txn.finish().unwrap();
+        txn.commit().unwrap();
+    }
+
+    #[test]
+    #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))]
+    fn ephemeral_mode_is_immutable() {
+        let tmp = tempfile::tempdir().unwrap();
+        {
+            let state = AppState::new(&config_at(tmp.path(), false, false)).unwrap();
+            write_record(&state, "3kzbif5moe22m", b"v1");
+        }
+
+        let err = match AppState::new(&config_at(tmp.path(), true, false)) {
+            Ok(_) => panic!("permanent to ephemeral flip must be rejected"),
+            Err(err) => err,
+        };
+        let msg = format!("{err}");
+        assert!(msg.contains("ephemeral=false"), "{msg}");
+        assert!(msg.contains("HYDRANT_EPHEMERAL"), "{msg}");
+        assert!(msg.contains("offline migration"), "{msg}");
+
+        AppState::new(&config_at(tmp.path(), false, false)).unwrap();
+    }
+
+    #[test]
+    #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))]
+    fn ephemeral_to_permanent_flip_is_rejected() {
+        let tmp = tempfile::tempdir().unwrap();
+        {
+            let state = AppState::new(&config_at(tmp.path(), true, false)).unwrap();
+            write_record(&state, "3kzbif5moe22m", b"v1");
+        }
+
+        let err = match AppState::new(&config_at(tmp.path(), false, false)) {
+            Ok(_) => panic!("ephemeral to permanent flip must be rejected"),
+            Err(err) => err,
+        };
+        assert!(format!("{err}").contains("ephemeral=true"));
+
+        AppState::new(&config_at(tmp.path(), true, false)).unwrap();
+    }
+
+    #[test]
+    #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))]
+    fn empty_database_still_records_its_ephemeral_identity() {
+        let tmp = tempfile::tempdir().unwrap();
+        AppState::new(&config_at(tmp.path(), true, false)).unwrap();
+
+        assert!(AppState::new(&config_at(tmp.path(), false, false)).is_err());
+    }
+
+    #[test]
+    #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))]
+    fn markerless_inline_events_prevent_wrong_permanent_adoption() {
+        let tmp = tempfile::tempdir().unwrap();
+        {
+            let state = AppState::new(&config_at(tmp.path(), true, false)).unwrap();
+            write_record(&state, "3kzbif5moe22m", b"v1");
+            assert!(state.db.indexer.records.is_empty().unwrap());
+            state.db.filter.remove(STORAGE_MODE_KEY).unwrap();
+            state.db.persist().unwrap();
+        }
+
+        let err = match AppState::new(&config_at(tmp.path(), false, false)) {
+            Ok(_) => panic!("inline v9 events must identify ephemeral mode"),
+            Err(err) => err,
+        };
+        assert!(format!("{err}").contains("imply ephemeral=true"));
+    }
+
+    #[test]
+    #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))]
+    fn markerless_pointer_events_prevent_wrong_ephemeral_adoption() {
+        let tmp = tempfile::tempdir().unwrap();
+        {
+            let state = AppState::new(&config_at(tmp.path(), false, false)).unwrap();
+            write_record(&state, "3kzbif5moe22m", b"v1");
+            state.db.filter.remove(STORAGE_MODE_KEY).unwrap();
+            state.db.persist().unwrap();
+        }
+
+        let err = match AppState::new(&config_at(tmp.path(), true, false)) {
+            Ok(_) => panic!("pointer v9 events must identify permanent mode"),
+            Err(err) => err,
+        };
+        assert!(format!("{err}").contains("imply ephemeral=false"));
+    }
+
+    #[test]
+    #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))]
+    fn marked_ephemeral_database_rejects_unshipped_record_heads() {
+        let tmp = tempfile::tempdir().unwrap();
+        {
+            let state = AppState::new(&config_at(tmp.path(), true, false)).unwrap();
+            let key = crate::db::keys::record_key(
+                &did(),
+                "app.bsky.feed.post",
+                &DbRkey::new("3kzbif5moe22m"),
+            );
+            state.db.indexer.records.insert(key, b"stale head").unwrap();
+            state.db.persist().unwrap();
+        }
+
+        let err = match AppState::new(&config_at(tmp.path(), true, false)) {
+            Ok(_) => panic!("event-only ephemeral mode must reject persisted heads"),
+            Err(err) => err,
+        };
+        assert!(format!("{err}").contains("unsupported pre-release"));
+    }
+
+    #[test]
+    #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))]
+    fn rejected_layout_open_cannot_arm_history_retention() {
+        let tmp = tempfile::tempdir().unwrap();
+        let record_key = crate::db::keys::record_key(
+            &did(),
+            "app.bsky.feed.post",
+            &DbRkey::new("3kzbif5moe22m"),
+        );
+        let history_key =
+            crate::db::keys::history_key(&record_key, &DbTid::new_from_bytes(1_u64.to_be_bytes()));
+
+        {
+            let state = AppState::new(&config_at(tmp.path(), false, false)).unwrap();
+            let mut batch = state.db.inner.batch();
+            state
+                .db
+                .indexer
+                .stage_history(&mut batch, &history_key, b"must survive".as_slice());
+            batch.commit().unwrap();
+            state.db.indexer.history.rotate_memtable_and_wait().unwrap();
+            state.db.persist().unwrap();
+        }
+
+        let mut wrong = config_at(tmp.path(), false, true);
+        wrong.history_ttl = Some(std::time::Duration::from_secs(1));
+        assert!(AppState::new(&wrong).is_err());
+
+        let state = AppState::new(&config_at(tmp.path(), false, false)).unwrap();
+        state.db.indexer.history.major_compact().unwrap();
+        assert_eq!(
+            state
+                .db
+                .indexer
+                .history
+                .get(history_key)
+                .unwrap()
+                .as_deref(),
+            Some(b"must survive".as_slice())
+        );
+    }
+
+    #[test]
+    fn links_only_flip_is_rejected_actionably() {
+        let tmp = tempfile::tempdir().unwrap();
+        {
+            let state = AppState::new(&config_at(tmp.path(), false, true)).unwrap();
+            write_record(&state, "3kzbif5moe22m", b"v1");
+        }
+
+        let err = match AppState::new(&config_at(tmp.path(), false, false)) {
+            Ok(_) => panic!("links-only flip must be rejected"),
+            Err(err) => err,
+        };
+        let msg = format!("{err}");
+        assert!(
+            msg.contains("not convertible in place"),
+            "unexpected error: {msg}"
+        );
+        assert!(msg.contains("only_index_links"), "unexpected error: {msg}");
+    }
+
+    #[test]
+    fn missing_marker_does_not_bless_a_links_only_database_as_body_storage() {
+        let tmp = tempfile::tempdir().unwrap();
+        {
+            let state = AppState::new(&config_at(tmp.path(), false, true)).unwrap();
+            write_record(&state, "3kzbif5moe22m", b"v1");
+            state.db.filter.remove(STORAGE_MODE_KEY).unwrap();
+            state.db.persist().unwrap();
+        }
+
+        let err = match AppState::new(&config_at(tmp.path(), false, false)) {
+            Ok(_) => panic!("pre-marker links-only mismatch must be rejected"),
+            Err(err) => err,
+        };
+        let msg = format!("{err}");
+        assert!(msg.contains("imply only_index_links=true"), "{msg}");
+    }
+
+    #[test]
+    fn missing_marker_does_not_bless_body_storage_as_links_only() {
+        let tmp = tempfile::tempdir().unwrap();
+        {
+            let state = AppState::new(&config_at(tmp.path(), false, false)).unwrap();
+            write_record(&state, "3kzbif5moe22m", b"v1");
+            state.db.filter.remove(STORAGE_MODE_KEY).unwrap();
+            state.db.persist().unwrap();
+        }
+
+        let err = match AppState::new(&config_at(tmp.path(), false, true)) {
+            Ok(_) => panic!("pre-marker body mismatch must be rejected"),
+            Err(err) => err,
+        };
+        let msg = format!("{err}");
+        assert!(msg.contains("imply only_index_links=false"), "{msg}");
+    }
+}
diff --git a/src/db/migration/v10/events.rs b/src/db/migration/v10/events.rs
--- a/src/db/migration/v10/events.rs
+++ b/src/db/migration/v10/events.rs
@@ -197,6 +197,8 @@
     use super::*;
     use crate::config::Config;
     use crate::db::types::{DbAction, DbRkey, DbTid, TrimmedDid};
+    #[cfg(not(feature = "backlinks"))]
+    use crate::state::AppState;
     use jacquard_common::types::string::{Did, Tid};
     use jacquard_common::{CowStr, IntoStatic};
     use miette::IntoDiagnostic;
@@ -680,4 +682,159 @@
         Ok(())
     }
 
+    #[test]
+    #[cfg(not(feature = "backlinks"))]
+    fn v9_ephemeral_database_upgrades_replays_and_expires() -> Result<()> {
+        let tmp = tempfile::tempdir().into_diagnostic()?;
+        let mut cfg = config(tmp.path());
+        cfg.ephemeral = true;
+        cfg.ephemeral_ttl = std::time::Duration::from_secs(60 * 60);
+        let legacy_body = body("legacy inline ephemeral");
+        let legacy_id = 7_u64;
+
+        // physical v9 ephemeral shape: replay bodies live inline in events,
+        // there are no record heads, and no storage-mode marker existed yet.
+        {
+            let db = Db::open(&cfg)?;
+            let mut batch = db.inner.batch();
+            stage_event(
+                &db,
+                &mut batch,
+                legacy_id,
+                &event(StoredData::Block(bytes::Bytes::copy_from_slice(
+                    &legacy_body,
+                ))),
+            );
+            let blocks = super::super::super::legacy_blocks_for_test(&db)?;
+            batch.insert(&blocks, b"legacy sentinel", b"legacy block");
+            batch.remove(&db.filter, crate::db::storage_mode::STORAGE_MODE_KEY);
+            batch.commit().into_diagnostic()?;
+            rewind_to_v9(&db)?;
+        }
+
+        let replay_body = |state: &AppState, id: u64| -> Result<Option<serde_json::Value>> {
+            let Some(bytes) = state
+                .db
+                .stream
+                .events
+                .get(keys::event_key(id))
+                .into_diagnostic()?
+            else {
+                return Ok(None);
+            };
+            let stored: StoredEvent<'_> = rmp_serde::from_slice(&bytes).into_diagnostic()?;
+            let event = crate::control::stream::indexer::stored_to_event(state, id, stored, None)
+                .ok_or_else(|| miette::miette!("stored event did not inflate"))?;
+            let raw = event
+                .record
+                .and_then(|record| record.record)
+                .ok_or_else(|| miette::miette!("stored event lost its record body"))?;
+            serde_json::from_str(raw.get()).into_diagnostic().map(Some)
+        };
+        let expected_legacy = serde_json::json!({
+            "$type": COLLECTION,
+            "text": "legacy inline ephemeral",
+        });
+
+        {
+            let state = AppState::new(&cfg)?;
+            assert!(!state.db.inner.keyspace_exists("blocks"));
+            assert!(state.db.indexer.records.is_empty().into_diagnostic()?);
+            assert_eq!(
+                replay_body(&state, legacy_id)?,
+                Some(expected_legacy.clone())
+            );
+            assert!(state.db.stream.event_bodies.is_empty().into_diagnostic()?);
+
+            // an ordinary tick cannot prune an event until a watermark has
+            // actually aged through the configured window.
+            crate::db::ephemeral::ephemeral_ttl_tick(&state.db, &cfg.ephemeral_ttl)?;
+            assert_eq!(
+                replay_body(&state, legacy_id)?,
+                Some(expected_legacy.clone())
+            );
+
+            state
+                .db
+                .stream
+                .events
+                .rotate_memtable_and_wait()
+                .into_diagnostic()?;
+            state.db.stream.events.major_compact().into_diagnostic()?;
+            state.db.persist()?;
+        }
+
+        let state = AppState::new(&cfg)?;
+        assert_eq!(replay_body(&state, legacy_id)?, Some(expected_legacy));
+
+        // once a v9 event's real retention window has elapsed, its inline body
+        // disappears atomically with the event.
+        let old_ts =
+            (chrono::Utc::now().timestamp() as u64).saturating_sub(cfg.ephemeral_ttl.as_secs() + 1);
+        state
+            .db
+            .cursors
+            .insert(
+                keys::event_watermark_key(old_ts),
+                (legacy_id + 1).to_be_bytes(),
+            )
+            .into_diagnostic()?;
+        crate::db::ephemeral::ephemeral_ttl_tick(&state.db, &cfg.ephemeral_ttl)?;
+        assert_eq!(replay_body(&state, legacy_id)?, None);
+        assert!(state.db.stream.event_bodies.is_empty().into_diagnostic()?);
+
+        // new writes stay event-only too: no head, history, redaction, or
+        // compatibility archive can retain the body past event TTL.
+        let new_body = bytes::Bytes::from(body("new ephemeral event"));
+        let new_cid = jacquard_repo::mst::util::compute_cid(&new_body).into_diagnostic()?;
+        let new_rkey = DbRkey::new("3kzbif5mof33m");
+        let new_key = keys::record_key(&did(), COLLECTION, &new_rkey);
+        let repo = did();
+        let mut txn = crate::db::Txn::new(&state.db);
+        let mut records = txn.records(&state, &DbTid::from(&Tid::now_0()), &repo)?;
+        records.put_record(COLLECTION, &new_rkey, &new_cid, &new_body, DbAction::Create)?;
+        records.finish()?;
+        txn.commit()?;
+
+        assert!(
+            state
+                .db
+                .indexer
+                .record(&new_key)
+                .into_diagnostic()?
+                .is_none()
+        );
+        assert!(state.db.indexer.history.is_empty().into_diagnostic()?);
+        assert!(state.db.indexer.redactions.is_empty().into_diagnostic()?);
+        assert!(state.db.stream.event_bodies.is_empty().into_diagnostic()?);
+
+        let new_id = legacy_id + 1;
+        let stored = state
+            .db
+            .stream
+            .events
+            .get(keys::event_key(new_id))
+            .into_diagnostic()?
+            .expect("new ephemeral event must be durable");
+        let stored: StoredEvent<'_> = rmp_serde::from_slice(&stored).into_diagnostic()?;
+        assert!(matches!(stored.data, StoredData::Block(found) if found == new_body));
+        assert_eq!(
+            replay_body(&state, new_id)?,
+            Some(serde_json::json!({
+                "$type": COLLECTION,
+                "text": "new ephemeral event",
+            }))
+        );
+
+        drop(state);
+        let mut permanent_cfg = cfg.clone();
+        permanent_cfg.ephemeral = false;
+        let err = match AppState::new(&permanent_cfg) {
+            Ok(_) => panic!("migrated ephemeral database must reject permanent mode"),
+            Err(err) => err,
+        };
+        assert!(format!("{err}").contains("ephemeral=true"));
+
+        Ok(())
+    }
 }