fix(station): stop unlistable cores from holding tape retention open

crates/moon-core/src/market/tape_recorder/catchup.rs

@@ -121,6 +121,29 @@ pub(super) fn listings_complete(generation: u64) -> bool {
         .clone();
     cores.is_some_and(|cores| all_listed(&cores, generation))
 }
+
+/// Required cores whose listing has not completed for `generation` — what the station log names
+/// when retention gives up waiting, since that barrier decides whether a pass deletes anything.
+/// `None` while the session has not published its cores yet, which holds the barrier as well.
+pub(super) fn unlisted(generation: u64) -> Option<Vec<u64>> {
+    let cores = REQUIRED_CORES
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .clone()?;
+    let listings = LISTINGS
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    Some(
+        cores
+            .into_iter()
+            .filter(|core| {
+                listings
+                    .get(core)
+                    .is_none_or(|entry| entry.1 != Some(generation))
+            })
+            .collect(),
+    )
+}
 static STATUS: Mutex<TapeCatchup> = Mutex::new(TapeCatchup {
     running: false,
     auto: None,
@@ -267,6 +290,21 @@ pub fn listed(core: u64, generation: u64) {
     super::send(Cmd::Listed { core, generation });
 }
 
+/// Settle at once the listing of a core that can attribute nothing: its platform is not in the
+/// exchange directory, so no report row can name a venue and the listing would protect no
+/// market. Left unlisted, such a core held [`listings_complete`] open forever, and every purge
+/// then kept all tape of every market — the age limit and the emulator choice alike. Applied
+/// directly: the queued [`listed`] only orders completion behind submitted pages, and there are
+/// none.
+///
+/// The cost, chosen by LinKvo on 2026-10-16: tape on such a core's own markets goes by age
+/// alone. Its trades are never known, so one straddling the retention edge loses its older part
+/// and its emulator tape stays until it ages out — as for a live close on an unknown venue.
+pub fn listed_empty(core: u64, generation: u64) {
+    listing_started(core, generation);
+    listing_applied(core, generation);
+}
+
 /// Apply a successful completion only when its retention generation is still current.
 pub(super) fn listing_applied(core: u64, generation: u64) {
     let _status = STATUS

crates/moon-core/src/market/tape_recorder/catchup/state.rs

@@ -141,6 +141,17 @@ impl RecorderCatchup {
                 };
                 let complete = super::listings_complete(generation);
                 if !self.active && (complete || now.saturating_sub(since) >= 600_000) {
+                    if !complete {
+                        // This pass keeps all tape of every market; say which cores held it.
+                        let cores = super::unlisted(generation).map_or_else(
+                            || "core list not received yet".to_string(),
+                            |cores| format!("cores not listed: {cores:?}"),
+                        );
+                        super::super::line(&format!(
+                            "retention: listings incomplete after 10 min, nothing is deleted \
+                             this pass; {cores}"
+                        ));
+                    }
                     self.retention.hour = Some(now.div_euclid(HOUR_MS));
                     status = Some(TapePurge::Running { done: 0, total: 0 });
                     Phase::Inventory {

crates/moon-core/src/market/tape_recorder/catchup/state/tests.rs

@@ -464,6 +464,37 @@ fn retention_drops_tape_that_ages_out() {
     );
 }
 
+/// A required core on a platform the exchange directory does not know can never list; once its
+/// empty listing is settled, the barrier closes and aged tape goes. Before, every pass fell back
+/// to keeping all tape after ten minutes, and a station's file never shrank.
+#[test]
+fn a_venue_less_core_does_not_hold_aged_tape() {
+    let _lock = SWITCH_LOCK.lock().unwrap();
+    let _restore = Restore::new();
+    let _settings = RetentionSettings::new();
+    catchup::set_keep(1, true);
+    let core = 70_001;
+    catchup::require_listings(vec![core]);
+    catchup::listed_empty(core, catchup::relist_gen());
+    let cache = TradeCache::open_evictable(":memory:".into()).unwrap();
+    cache.insert(
+        "synthetic",
+        "SYNUSDT",
+        NOW - 3 * DAY,
+        NOW - 2 * DAY,
+        vec![],
+        crate::market::trade_replay::tick_tiles::TileSource::Core,
+    );
+    let mut state = RecorderCatchup::default();
+    settle(&mut state, &cache, NOW);
+    assert!(
+        cache
+            .held_spans("synthetic", "SYNUSDT", i64::MIN, i64::MAX)
+            .unwrap()
+            .is_empty()
+    );
+}
+
 /// Without a deferred request, forever retention would never revisit an emulator purge vetoed by an offline report.
 #[test]
 fn incomplete_listing_defers_emulator_purge() {

crates/moon-core/src/market/tape_recorder/catchup/tests.rs

@@ -309,6 +309,19 @@ fn two_reloads_during_one_listing_count_only_the_last() {
     take_purge_wanted();
 }
 
+/// A core whose platform is outside the exchange directory lists nothing; its listing must close
+/// the barrier at once, or every purge keeps all tape of every market forever.
+#[test]
+fn an_empty_listing_settles_the_barrier_at_once() {
+    let _guard = SWITCH_LOCK.lock().unwrap();
+    let core = 9_000_011;
+    let generation = relist_gen();
+    assert!(!all_listed(&[core], generation));
+    listed_empty(core, generation);
+    assert!(all_listed(&[core], generation));
+    assert_eq!(listing_generation(core), Some(generation));
+}
+
 /// Synthetic closed positions share no production core or market identity.
 fn trade(id: i64, brand: Brand, kind: MarketKind, age_ms: i64) -> CatchupTrade {
     let close_ms = NOW - age_ms;

crates/moon-core/src/session/lifecycle.rs

@@ -796,7 +796,15 @@ impl SessionManager {
             }
         }
         if crate::market::tape_recorder::catchup::enabled() {
-            crate::market::tape_recorder::catchup::require_listings(self.config_order.clone());
+            // Every session, connected or not — but not a server switched off in the config: it
+            // has no session, could never list, and would hold the retention barrier forever.
+            // The cost, chosen by LinKvo on 2026-10-16: its trades are not known to retention,
+            // so tape under them goes by age alone — one straddling the edge loses its older
+            // part, and on a shared market it gets no shelter from another core's emulator or
+            // tuner-unusable subtraction.
+            crate::market::tape_recorder::catchup::require_listings(
+                self.sessions.iter().map(|sess| sess.id).collect(),
+            );
             let now = Instant::now();
             let generation = crate::market::tape_recorder::catchup::relist_gen();
             let cores: Vec<_> = self
@@ -831,9 +839,14 @@ impl SessionManager {
             return false;
         };
         // The catch-up fetches from a venue's public REST: a venue the directory does not know
-        // has no route, so its core is not listed — as before its address resolved at all.
+        // has no route, so its core is not listed. Its listing still counts as complete — it
+        // could protect nothing — or the retention barrier would wait on it forever.
         let Some(venue) = address.venue else {
-            return false;
+            crate::market::tape_recorder::catchup::listed_empty(
+                core,
+                crate::market::tape_recorder::catchup::relist_gen(),
+            );
+            return true;
         };
         let catalog_ready = self.market_source.tradable_coins(core).is_some();
         if !catalog_ready {