implements PLN-0128 asset pipeline telemetry tests

crates/console/prometeu-drivers/src/asset.rs

@@ -34,6 +34,134 @@ type HandleTable = HashMap<HandleId, LoadHandleInfo>;
 type TargetHandleTable = HashMap<SlotRef, HandleId>;
 type TargetGenerationTable = HashMap<SlotRef, u64>;
 
+const ASSET_PROGRESS_IDLE: u16 = 0;
+const ASSET_PROGRESS_LOADING: u16 = 1_000;
+const ASSET_PROGRESS_DONE: u16 = 10_000;
+const ASSET_TELEMETRY_SAMPLE_WINDOW: usize = 32;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub struct AssetLatencyPercentiles {
+    pub p50_micros: u64,
+    pub p95_micros: u64,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
+pub struct AssetPipelineTelemetry {
+    pub submitted_requests: u64,
+    pub started_jobs: u64,
+    pub completed_jobs: u64,
+    pub failed_jobs: u64,
+    pub canceled_requests: u64,
+    pub superseded_requests: u64,
+    pub stale_results_discarded: u64,
+    pub current_backlog_depth: usize,
+    pub max_backlog_depth: usize,
+    pub active_progress: u16,
+    pub glyph_latency: AssetLatencyPercentiles,
+    pub sound_latency: AssetLatencyPercentiles,
+    pub scene_latency: AssetLatencyPercentiles,
+    pub last_asset_id: Option<AssetId>,
+    pub last_asset_latency: AssetLatencyPercentiles,
+}
+
+#[derive(Default)]
+struct AssetPipelineTelemetryState {
+    snapshot: AssetPipelineTelemetry,
+    bank_samples: HashMap<BankType, VecDeque<u64>>,
+    asset_samples: HashMap<AssetId, VecDeque<u64>>,
+}
+
+impl AssetPipelineTelemetryState {
+    fn record_submitted(&mut self, backlog_depth: usize) {
+        self.snapshot.submitted_requests += 1;
+        self.snapshot.current_backlog_depth = backlog_depth;
+        self.snapshot.max_backlog_depth = self.snapshot.max_backlog_depth.max(backlog_depth);
+    }
+
+    fn record_started(&mut self, backlog_depth: usize) {
+        self.snapshot.started_jobs += 1;
+        self.snapshot.current_backlog_depth = backlog_depth;
+        self.snapshot.active_progress = ASSET_PROGRESS_LOADING;
+    }
+
+    fn record_completed(&mut self, asset_id: AssetId, bank_type: BankType, duration_micros: u64) {
+        self.snapshot.completed_jobs += 1;
+        self.snapshot.active_progress = ASSET_PROGRESS_IDLE;
+        self.record_latency_sample(asset_id, bank_type, duration_micros);
+    }
+
+    fn record_failed(&mut self, asset_id: AssetId, bank_type: BankType, duration_micros: u64) {
+        self.snapshot.failed_jobs += 1;
+        self.snapshot.active_progress = ASSET_PROGRESS_IDLE;
+        self.record_latency_sample(asset_id, bank_type, duration_micros);
+    }
+
+    fn record_canceled(&mut self) {
+        self.snapshot.canceled_requests += 1;
+    }
+
+    fn record_superseded(&mut self, count: usize) {
+        self.snapshot.superseded_requests += count as u64;
+    }
+
+    fn record_stale_discard(&mut self) {
+        self.snapshot.stale_results_discarded += 1;
+        self.snapshot.superseded_requests += 1;
+        self.snapshot.active_progress = ASSET_PROGRESS_IDLE;
+    }
+
+    fn record_latency_sample(
+        &mut self,
+        asset_id: AssetId,
+        bank_type: BankType,
+        duration_micros: u64,
+    ) {
+        push_sample(
+            self.bank_samples.entry(bank_type).or_default(),
+            duration_micros,
+            ASSET_TELEMETRY_SAMPLE_WINDOW,
+        );
+        push_sample(
+            self.asset_samples.entry(asset_id).or_default(),
+            duration_micros,
+            ASSET_TELEMETRY_SAMPLE_WINDOW,
+        );
+        self.snapshot.glyph_latency = percentile_snapshot(self.bank_samples.get(&BankType::GLYPH));
+        self.snapshot.sound_latency = percentile_snapshot(self.bank_samples.get(&BankType::SOUNDS));
+        self.snapshot.scene_latency = percentile_snapshot(self.bank_samples.get(&BankType::SCENE));
+        self.snapshot.last_asset_id = Some(asset_id);
+        self.snapshot.last_asset_latency = percentile_snapshot(self.asset_samples.get(&asset_id));
+    }
+}
+
+fn push_sample(samples: &mut VecDeque<u64>, value: u64, window: usize) {
+    samples.push_back(value);
+    while samples.len() > window {
+        samples.pop_front();
+    }
+}
+
+fn percentile_snapshot(samples: Option<&VecDeque<u64>>) -> AssetLatencyPercentiles {
+    let Some(samples) = samples else {
+        return AssetLatencyPercentiles::default();
+    };
+    if samples.is_empty() {
+        return AssetLatencyPercentiles::default();
+    }
+
+    let mut sorted = samples.iter().copied().collect::<Vec<_>>();
+    sorted.sort_unstable();
+    AssetLatencyPercentiles {
+        p50_micros: percentile_value(&sorted, 50),
+        p95_micros: percentile_value(&sorted, 95),
+    }
+}
+
+fn percentile_value(sorted: &[u64], percentile: usize) -> u64 {
+    let index = ((sorted.len() - 1) * percentile).div_ceil(100);
+    sorted[index]
+}
+
 #[derive(Clone, Default)]
 pub struct GlyphAssetSlotIndex {
     slots_by_asset_id: Arc<RwLock<HashMap<AssetId, usize>>>,
@@ -188,6 +316,7 @@ pub struct AssetManager {
     target_generations: Arc<RwLock<TargetGenerationTable>>,
     next_handle_id: Mutex<HandleId>,
     assets_data: Arc<RwLock<AssetsPayloadSource>>,
+    pipeline_telemetry: Arc<Mutex<AssetPipelineTelemetryState>>,
 
     /// Narrow hardware interfaces
     gfx_installer: Arc<dyn GlyphBankPoolInstaller>,
@@ -218,6 +347,7 @@ struct LoadHandleInfo {
     slot: SlotRef,
     status: LoadStatus,
     request_generation: u64,
+    progress: u16,
 }
 
 #[derive(Clone)]
@@ -243,14 +373,17 @@ struct AssetLoadQueue {
 }
 
 impl AssetLoadQueue {
-    fn submit(&self, job: AssetLoadJob) {
+    fn submit(&self, job: AssetLoadJob) -> usize {
         let mut state = self.state.lock().unwrap();
         if state.shutdown {
-            return;
+            return 0;
         }
+        let before_len = state.pending.len();
         state.pending.retain(|pending| pending.slot != job.slot);
+        let removed = before_len - state.pending.len();
         state.pending.push_back(job);
         self.ready.notify_one();
+        removed
     }
 
     fn next_job(&self) -> Option<AssetLoadJob> {
@@ -334,6 +467,7 @@ impl AssetLoadWorker {
         handles: Arc<RwLock<HandleTable>>,
         target_generations: Arc<RwLock<TargetGenerationTable>>,
         assets_data: Arc<RwLock<AssetsPayloadSource>>,
+        pipeline_telemetry: Arc<Mutex<AssetPipelineTelemetryState>>,
         gfx_policy: BankPolicy<GlyphBank>,
         sound_policy: BankPolicy<SoundBank>,
         scene_policy: BankPolicy<SceneBank>,
@@ -347,6 +481,7 @@ impl AssetLoadWorker {
                     &handles,
                     &target_generations,
                     &assets_data,
+                    &pipeline_telemetry,
                     &gfx_policy,
                     &sound_policy,
                     &scene_policy,
@@ -358,8 +493,8 @@ impl AssetLoadWorker {
         Self { queue, join_handle: Some(join_handle) }
     }
 
-    fn submit(&self, job: AssetLoadJob) {
-        self.queue.submit(job);
+    fn submit_counting_superseded(&self, job: AssetLoadJob) -> usize {
+        self.queue.submit(job)
     }
 
     fn clear_pending(&self) {
@@ -533,6 +668,7 @@ impl AssetManager {
         let handles = Arc::new(RwLock::new(HashMap::new()));
         let target_generations = Arc::new(RwLock::new(HashMap::new()));
         let assets_data = Arc::new(RwLock::new(assets_data));
+        let pipeline_telemetry = Arc::new(Mutex::new(AssetPipelineTelemetryState::default()));
         let gfx_policy = BankPolicy::new();
         let sound_policy = BankPolicy::new();
         let scene_policy = BankPolicy::new();
@@ -540,6 +676,7 @@ impl AssetManager {
             Arc::clone(&handles),
             Arc::clone(&target_generations),
             Arc::clone(&assets_data),
+            Arc::clone(&pipeline_telemetry),
             gfx_policy.clone(),
             sound_policy.clone(),
             scene_policy.clone(),
@@ -562,6 +699,7 @@ impl AssetManager {
             target_generations,
             next_handle_id: Mutex::new(1),
             assets_data,
+            pipeline_telemetry,
             load_worker,
             pending_commits: Mutex::new(Vec::new()),
         }
@@ -769,8 +907,11 @@ impl AssetManager {
                     slot,
                     status: LoadStatus::READY,
                     request_generation,
+                    progress: ASSET_PROGRESS_DONE,
                 },
             );
+            self.pipeline_telemetry.lock().unwrap().record_submitted(0);
+            self.pipeline_telemetry.lock().unwrap().record_completed(asset_id, entry.bank_type, 0);
             return Ok(handle_id);
         }
 
@@ -782,16 +923,21 @@ impl AssetManager {
                 slot,
                 status: LoadStatus::PENDING,
                 request_generation,
+                progress: ASSET_PROGRESS_IDLE,
             },
         );
 
-        self.load_worker.submit(AssetLoadJob {
+        let superseded = self.load_worker.submit_counting_superseded(AssetLoadJob {
             handle_id,
             asset_id,
             slot,
             request_generation,
             entry,
         });
+        let backlog_depth = self.load_worker.pending_count();
+        let mut telemetry = self.pipeline_telemetry.lock().unwrap();
+        telemetry.record_submitted(backlog_depth);
+        telemetry.record_superseded(superseded);
 
         Ok(handle_id)
     }
@@ -801,12 +947,14 @@ impl AssetManager {
         handles: &Arc<RwLock<HandleTable>>,
         target_generations: &Arc<RwLock<TargetGenerationTable>>,
         assets_data: &Arc<RwLock<AssetsPayloadSource>>,
+        pipeline_telemetry: &Arc<Mutex<AssetPipelineTelemetryState>>,
         gfx_policy: &BankPolicy<GlyphBank>,
         sound_policy: &BankPolicy<SoundBank>,
         scene_policy: &BankPolicy<SceneBank>,
     ) {
         if !Self::is_current_target_generation(target_generations, job.slot, job.request_generation)
         {
+            pipeline_telemetry.lock().unwrap().record_stale_discard();
             return;
         }
 
@@ -821,7 +969,10 @@ impl AssetManager {
                 return;
             }
             handle.status = LoadStatus::LOADING;
+            handle.progress = ASSET_PROGRESS_LOADING;
         }
+        let started_at = Instant::now();
+        pipeline_telemetry.lock().unwrap().record_started(0);
 
         match job.entry.bank_type {
             BankType::GLYPH => {
@@ -833,6 +984,7 @@ impl AssetManager {
                             job.slot,
                             job.request_generation,
                         ) {
+                            pipeline_telemetry.lock().unwrap().record_stale_discard();
                             return;
                         }
                         let bank_arc = Arc::new(tilebank);
@@ -846,6 +998,7 @@ impl AssetManager {
                             job.slot,
                             job.request_generation,
                         ) {
+                            pipeline_telemetry.lock().unwrap().record_stale_discard();
                             return;
                         }
                         gfx_policy.stage(
@@ -854,8 +1007,22 @@ impl AssetManager {
                             job.entry.decoded_size as usize,
                         );
                         Self::complete_load_job(handles, &job, LoadStatus::READY);
+                        Self::record_load_job_finished(
+                            pipeline_telemetry,
+                            &job,
+                            started_at,
+                            LoadStatus::READY,
+                        );
+                    }
+                    Err(_) => {
+                        Self::complete_load_job(handles, &job, LoadStatus::ERROR);
+                        Self::record_load_job_finished(
+                            pipeline_telemetry,
+                            &job,
+                            started_at,
+                            LoadStatus::ERROR,
+                        );
                     }
-                    Err(_) => Self::complete_load_job(handles, &job, LoadStatus::ERROR),
                 }
             }
             BankType::SOUNDS => {
@@ -867,6 +1034,7 @@ impl AssetManager {
                             job.slot,
                             job.request_generation,
                         ) {
+                            pipeline_telemetry.lock().unwrap().record_stale_discard();
                             return;
                         }
                         let bank_arc = Arc::new(soundbank);
@@ -880,6 +1048,7 @@ impl AssetManager {
                             job.slot,
                             job.request_generation,
                         ) {
+                            pipeline_telemetry.lock().unwrap().record_stale_discard();
                             return;
                         }
                         sound_policy.stage(
@@ -888,8 +1057,22 @@ impl AssetManager {
                             job.entry.decoded_size as usize,
                         );
                         Self::complete_load_job(handles, &job, LoadStatus::READY);
+                        Self::record_load_job_finished(
+                            pipeline_telemetry,
+                            &job,
+                            started_at,
+                            LoadStatus::READY,
+                        );
+                    }
+                    Err(_) => {
+                        Self::complete_load_job(handles, &job, LoadStatus::ERROR);
+                        Self::record_load_job_finished(
+                            pipeline_telemetry,
+                            &job,
+                            started_at,
+                            LoadStatus::ERROR,
+                        );
                     }
-                    Err(_) => Self::complete_load_job(handles, &job, LoadStatus::ERROR),
                 }
             }
             BankType::SCENE => {
@@ -901,6 +1084,7 @@ impl AssetManager {
                             job.slot,
                             job.request_generation,
                         ) {
+                            pipeline_telemetry.lock().unwrap().record_stale_discard();
                             return;
                         }
                         let bank_arc = Arc::new(scenebank);
@@ -914,6 +1098,7 @@ impl AssetManager {
                             job.slot,
                             job.request_generation,
                         ) {
+                            pipeline_telemetry.lock().unwrap().record_stale_discard();
                             return;
                         }
                         scene_policy.stage(
@@ -922,13 +1107,46 @@ impl AssetManager {
                             job.entry.decoded_size as usize,
                         );
                         Self::complete_load_job(handles, &job, LoadStatus::READY);
+                        Self::record_load_job_finished(
+                            pipeline_telemetry,
+                            &job,
+                            started_at,
+                            LoadStatus::READY,
+                        );
+                    }
+                    Err(_) => {
+                        Self::complete_load_job(handles, &job, LoadStatus::ERROR);
+                        Self::record_load_job_finished(
+                            pipeline_telemetry,
+                            &job,
+                            started_at,
+                            LoadStatus::ERROR,
+                        );
                     }
-                    Err(_) => Self::complete_load_job(handles, &job, LoadStatus::ERROR),
                 }
             }
         }
     }
 
+    fn record_load_job_finished(
+        pipeline_telemetry: &Arc<Mutex<AssetPipelineTelemetryState>>,
+        job: &AssetLoadJob,
+        started_at: Instant,
+        status: LoadStatus,
+    ) {
+        let duration_micros = started_at.elapsed().as_micros() as u64;
+        let mut telemetry = pipeline_telemetry.lock().unwrap();
+        match status {
+            LoadStatus::READY => {
+                telemetry.record_completed(job.asset_id, job.entry.bank_type, duration_micros)
+            }
+            LoadStatus::ERROR => {
+                telemetry.record_failed(job.asset_id, job.entry.bank_type, duration_micros)
+            }
+            _ => {}
+        }
+    }
+
     fn complete_load_job(
         handles: &Arc<RwLock<HandleTable>>,
         job: &AssetLoadJob,
@@ -940,6 +1158,7 @@ impl AssetManager {
                 && handle.status == LoadStatus::LOADING
             {
                 handle.status = status;
+                handle.progress = ASSET_PROGRESS_DONE;
             }
         }
     }
@@ -1291,6 +1510,10 @@ impl AssetManager {
 
     pub fn backlog_info(&self) -> AssetBacklogInfo {
         let active = self.load_worker.active_job();
+        let active_progress = active
+            .as_ref()
+            .and_then(|job| self.handles.read().unwrap().get(&job.handle_id).map(|h| h.progress))
+            .unwrap_or(ASSET_PROGRESS_IDLE);
         AssetBacklogInfo {
             status: AssetOpStatus::Ok,
             pending_count: self.load_worker.pending_count() as u32,
@@ -1298,7 +1521,7 @@ impl AssetManager {
             active_asset_id: active.as_ref().map(|job| job.asset_id),
             active_bank_type: active.as_ref().map(|job| job.slot.asset_type),
             active_slot: active.as_ref().map(|job| job.slot.index),
-            active_progress: 0,
+            active_progress,
         }
     }
 
@@ -1317,7 +1540,13 @@ impl AssetManager {
                 status: AssetOpStatus::Ok,
                 state: LoadStatus::QUEUED,
                 position: position as u32,
-                progress: 0,
+                progress: self
+                    .handles
+                    .read()
+                    .unwrap()
+                    .get(&handle)
+                    .map(|h| h.progress)
+                    .unwrap_or(ASSET_PROGRESS_IDLE),
             };
         }
 
@@ -1326,7 +1555,13 @@ impl AssetManager {
                 status: AssetOpStatus::Ok,
                 state: LoadStatus::ACTIVE,
                 position: 0,
-                progress: 0,
+                progress: self
+                    .handles
+                    .read()
+                    .unwrap()
+                    .get(&handle)
+                    .map(|h| h.progress)
+                    .unwrap_or(ASSET_PROGRESS_LOADING),
             };
         }
 
@@ -1340,7 +1575,13 @@ impl AssetManager {
                 .map(|h| h.status)
                 .unwrap_or_else(|| self.idle_target_state(slot)),
             position: 0,
-            progress: 0,
+            progress: self
+                .handles
+                .read()
+                .unwrap()
+                .get(&handle)
+                .map(|h| h.progress)
+                .unwrap_or(ASSET_PROGRESS_IDLE),
         }
     }
 
@@ -1407,6 +1648,13 @@ impl AssetManager {
         }
     }
 
+    pub fn pipeline_telemetry(&self) -> AssetPipelineTelemetry {
+        let mut snapshot = self.pipeline_telemetry.lock().unwrap().snapshot.clone();
+        snapshot.current_backlog_depth = self.load_worker.pending_count();
+        snapshot.active_progress = self.backlog_info().active_progress;
+        snapshot
+    }
+
     pub fn commit(&self, handle: HandleId) -> AssetOpStatus {
         let mut handles_map = self.handles.write().unwrap();
         let Some(h) = handles_map.get_mut(&handle) else {
@@ -1436,6 +1684,7 @@ impl AssetManager {
                 match h.status {
                     LoadStatus::PENDING | LoadStatus::LOADING | LoadStatus::READY => {
                         h.status = LoadStatus::CANCELED;
+                        h.progress = ASSET_PROGRESS_DONE;
                         canceled_slot = Some(h.slot);
                     }
                     _ => {}
@@ -1444,6 +1693,7 @@ impl AssetManager {
         }
         if let Some(slot) = canceled_slot {
             self.bump_target_generation(slot);
+            self.pipeline_telemetry.lock().unwrap().record_canceled();
         }
         self.clear_staging_for_handle(handle);
         final_status
@@ -2089,6 +2339,26 @@ mod tests {
         assert_eq!(queue.pending_position(3), Some(3));
     }
 
+    #[test]
+    fn test_asset_backlog_queue_keeps_only_latest_request_per_target() {
+        let queue = AssetLoadQueue::default();
+        let entry = test_glyph_asset_entry("queued_glyphs", 16, 16);
+        let job = |handle_id: HandleId, request_generation: u64| AssetLoadJob {
+            handle_id,
+            asset_id: handle_id as AssetId,
+            slot: SlotRef::gfx(0),
+            request_generation,
+            entry: entry.clone(),
+        };
+
+        assert_eq!(queue.submit(job(1, 1)), 0);
+        assert_eq!(queue.submit(job(2, 2)), 1);
+
+        assert_eq!(queue.pending_count(), 1);
+        assert_eq!(queue.pending_position(1), None);
+        assert_eq!(queue.pending_position(2), Some(1));
+    }
+
     #[test]
     fn test_asset_target_status_exposes_empty_and_ready_slot_handle() {
         let banks = Arc::new(MemoryBanks::new());
@@ -2129,6 +2399,81 @@ mod tests {
         assert_eq!(ready.state, LoadStatus::READY);
     }
 
+    #[test]
+    fn test_asset_pipeline_progress_and_telemetry_snapshot_for_success() {
+        let banks = Arc::new(MemoryBanks::new());
+        let gfx_installer = Arc::clone(&banks) as Arc<dyn GlyphBankPoolInstaller>;
+        let sound_installer = Arc::clone(&banks) as Arc<dyn SoundBankPoolInstaller>;
+        let scene_installer = Arc::clone(&banks) as Arc<dyn SceneBankPoolInstaller>;
+
+        let data = test_glyph_asset_data();
+        let asset_entry = test_glyph_asset_entry("test_glyphs", 16, 16);
+
+        let am = AssetManager::new(
+            vec![asset_entry],
+            AssetsPayloadSource::from_bytes(data),
+            gfx_installer,
+            sound_installer,
+            scene_installer,
+        );
+
+        let empty = am.target_status(BankType::GLYPH, 0);
+        assert_eq!(empty.progress, ASSET_PROGRESS_IDLE);
+
+        let handle = am.load(0, 0).unwrap();
+        let start = Instant::now();
+        while am.status(handle) != LoadStatus::READY && start.elapsed().as_secs() < 5 {
+            thread::sleep(std::time::Duration::from_millis(10));
+        }
+
+        let position = am.backlog_position(handle);
+        assert_eq!(position.state, LoadStatus::READY);
+        assert_eq!(position.progress, ASSET_PROGRESS_DONE);
+
+        let telemetry = am.pipeline_telemetry();
+        assert_eq!(telemetry.submitted_requests, 1);
+        assert_eq!(telemetry.started_jobs, 1);
+        assert_eq!(telemetry.completed_jobs, 1);
+        assert_eq!(telemetry.failed_jobs, 0);
+        assert_eq!(telemetry.active_progress, ASSET_PROGRESS_IDLE);
+        assert_eq!(telemetry.last_asset_id, Some(0));
+    }
+
+    #[test]
+    fn test_asset_pipeline_progress_and_telemetry_snapshot_for_failure() {
+        let banks = Arc::new(MemoryBanks::new());
+        let gfx_installer = Arc::clone(&banks) as Arc<dyn GlyphBankPoolInstaller>;
+        let sound_installer = Arc::clone(&banks) as Arc<dyn SoundBankPoolInstaller>;
+        let scene_installer = Arc::clone(&banks) as Arc<dyn SceneBankPoolInstaller>;
+
+        let asset_entry = test_glyph_asset_entry("broken_glyphs", 16, 16);
+
+        let am = AssetManager::new(
+            vec![asset_entry],
+            AssetsPayloadSource::from_bytes(Vec::new()),
+            gfx_installer,
+            sound_installer,
+            scene_installer,
+        );
+
+        let handle = am.load(0, 0).unwrap();
+        let start = Instant::now();
+        while am.status(handle) != LoadStatus::ERROR && start.elapsed().as_secs() < 5 {
+            thread::sleep(std::time::Duration::from_millis(10));
+        }
+
+        let position = am.backlog_position(handle);
+        assert_eq!(position.state, LoadStatus::ERROR);
+        assert_eq!(position.progress, ASSET_PROGRESS_DONE);
+
+        let telemetry = am.pipeline_telemetry();
+        assert_eq!(telemetry.submitted_requests, 1);
+        assert_eq!(telemetry.started_jobs, 1);
+        assert_eq!(telemetry.completed_jobs, 0);
+        assert_eq!(telemetry.failed_jobs, 1);
+        assert_eq!(telemetry.last_asset_id, Some(0));
+    }
+
     #[test]
     fn test_asset_backlog_reorder_rejects_non_queued_request() {
         let banks = Arc::new(MemoryBanks::new());

crates/console/prometeu-hal/src/asset.rs

@@ -146,7 +146,7 @@ pub struct AssetBacklogInfo {
     pub active_asset_id: Option<AssetId>,
     pub active_bank_type: Option<BankType>,
     pub active_slot: Option<usize>,
-    pub active_progress: u8,
+    pub active_progress: u16,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -154,7 +154,7 @@ pub struct AssetBacklogPosition {
     pub status: AssetOpStatus,
     pub state: LoadStatus,
     pub position: u32,
-    pub progress: u8,
+    pub progress: u16,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -164,7 +164,7 @@ pub struct AssetTargetStatus {
     pub handle: HandleId,
     pub state: LoadStatus,
     pub position: u32,
-    pub progress: u8,
+    pub progress: u16,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

discussion/index.ndjson

@@ -18,7 +18,7 @@
 {"type":"discussion","id":"DSC-0006","status":"open","ticket":"vm-owned-random-service","title":"Agenda - VM-Owned Random Service","created_at":"2026-03-27","updated_at":"2026-03-27","tags":[],"agendas":[{"id":"AGD-0005","file":"AGD-0005-vm-owned-random-service.md","status":"open","created_at":"2026-03-27","updated_at":"2026-03-27"}],"decisions":[],"plans":[],"lessons":[]}
 {"type":"discussion","id":"DSC-0007","status":"open","ticket":"app-home-filesystem-surface-and-semantics","title":"Agenda - App Home Filesystem Surface and Semantics","created_at":"2026-03-27","updated_at":"2026-03-27","tags":[],"agendas":[{"id":"AGD-0006","file":"AGD-0006-app-home-filesystem-surface-and-semantics.md","status":"open","created_at":"2026-03-27","updated_at":"2026-03-27"}],"decisions":[],"plans":[],"lessons":[]}
 {"type":"discussion","id":"DSC-0008","status":"done","ticket":"perf-runtime-telemetry-hot-path","title":"Agenda - [PERF] Runtime Telemetry Hot Path","created_at":"2026-03-27","updated_at":"2026-06-04","tags":[],"agendas":[],"decisions":[],"plans":[],"lessons":[{"id":"LSN-0026","file":"discussion/lessons/DSC-0008-perf-runtime-telemetry-hot-path/LSN-0026-push-based-telemetry-model.md","status":"done","created_at":"2026-04-10","updated_at":"2026-04-10"}]}
-{"type":"discussion","id":"DSC-0009","status":"in_progress","ticket":"perf-async-background-work-lanes-for-assets-and-fs","title":"Agenda - [PERF] Async Background Work Lanes for Assets and FS","created_at":"2026-03-27","updated_at":"2026-06-28","tags":["perf","asset","fs","async","scheduler","runtime"],"agendas":[{"id":"AGD-0008","file":"AGD-0008-perf-async-background-work-lanes-for-assets-and-fs.md","status":"accepted","created_at":"2026-03-27","updated_at":"2026-06-28"}],"decisions":[{"id":"DEC-0034","file":"DEC-0034-async-work-lane-and-asset-backlog-contract.md","status":"accepted","created_at":"2026-06-28","updated_at":"2026-06-28","ref_agenda":"AGD-0008"}],"plans":[{"id":"PLN-0123","file":"PLN-0123-async-work-lane-specification-propagation.md","status":"done","created_at":"2026-06-28","updated_at":"2026-06-28","ref_decisions":["DEC-0034"]},{"id":"PLN-0124","file":"PLN-0124-async-work-lane-runtime-infrastructure.md","status":"done","created_at":"2026-06-28","updated_at":"2026-06-28","ref_decisions":["DEC-0034"]},{"id":"PLN-0125","file":"PLN-0125-asset-backlog-and-stable-slot-handles.md","status":"done","created_at":"2026-06-28","updated_at":"2026-06-28","ref_decisions":["DEC-0034"]},{"id":"PLN-0126","file":"PLN-0126-asset-backlog-public-api-and-status-surface.md","status":"done","created_at":"2026-06-28","updated_at":"2026-06-28","ref_decisions":["DEC-0034"]},{"id":"PLN-0127","file":"PLN-0127-async-lane-memcard-and-fs-integration-boundaries.md","status":"done","created_at":"2026-06-28","updated_at":"2026-06-28","ref_decisions":["DEC-0034"]},{"id":"PLN-0128","file":"PLN-0128-async-asset-pipeline-contract-tests-and-telemetry.md","status":"open","created_at":"2026-06-28","updated_at":"2026-06-28","ref_decisions":["DEC-0034"]}],"lessons":[]}
+{"type":"discussion","id":"DSC-0009","status":"in_progress","ticket":"perf-async-background-work-lanes-for-assets-and-fs","title":"Agenda - [PERF] Async Background Work Lanes for Assets and FS","created_at":"2026-03-27","updated_at":"2026-06-28","tags":["perf","asset","fs","async","scheduler","runtime"],"agendas":[{"id":"AGD-0008","file":"AGD-0008-perf-async-background-work-lanes-for-assets-and-fs.md","status":"accepted","created_at":"2026-03-27","updated_at":"2026-06-28"}],"decisions":[{"id":"DEC-0034","file":"DEC-0034-async-work-lane-and-asset-backlog-contract.md","status":"accepted","created_at":"2026-06-28","updated_at":"2026-06-28","ref_agenda":"AGD-0008"}],"plans":[{"id":"PLN-0123","file":"PLN-0123-async-work-lane-specification-propagation.md","status":"done","created_at":"2026-06-28","updated_at":"2026-06-28","ref_decisions":["DEC-0034"]},{"id":"PLN-0124","file":"PLN-0124-async-work-lane-runtime-infrastructure.md","status":"done","created_at":"2026-06-28","updated_at":"2026-06-28","ref_decisions":["DEC-0034"]},{"id":"PLN-0125","file":"PLN-0125-asset-backlog-and-stable-slot-handles.md","status":"done","created_at":"2026-06-28","updated_at":"2026-06-28","ref_decisions":["DEC-0034"]},{"id":"PLN-0126","file":"PLN-0126-asset-backlog-public-api-and-status-surface.md","status":"done","created_at":"2026-06-28","updated_at":"2026-06-28","ref_decisions":["DEC-0034"]},{"id":"PLN-0127","file":"PLN-0127-async-lane-memcard-and-fs-integration-boundaries.md","status":"done","created_at":"2026-06-28","updated_at":"2026-06-28","ref_decisions":["DEC-0034"]},{"id":"PLN-0128","file":"PLN-0128-async-asset-pipeline-contract-tests-and-telemetry.md","status":"done","created_at":"2026-06-28","updated_at":"2026-06-28","ref_decisions":["DEC-0034"]}],"lessons":[]}
 {"type":"discussion","id":"DSC-0010","status":"done","ticket":"perf-host-desktop-frame-pacing-and-presentation","title":"Agenda - [PERF] Host Desktop Frame Pacing and Presentation","created_at":"2026-03-27","updated_at":"2026-04-20","tags":[],"agendas":[],"decisions":[],"plans":[],"lessons":[{"id":"LSN-0036","file":"discussion/lessons/DSC-0010-perf-host-desktop-frame-pacing-and-presentation/LSN-0036-frame-publication-and-host-invalidation-must-be-separate.md","status":"done","created_at":"2026-04-20","updated_at":"2026-04-20"}]}
 {"type":"discussion","id":"DSC-0011","status":"abandoned","ticket":"perf-gfx-render-pipeline-and-dirty-regions","title":"Agenda - [PERF] GFX Render Pipeline and Dirty Regions","created_at":"2026-03-27","updated_at":"2026-06-28","tags":[],"agendas":[{"id":"AGD-0010","file":"AGD-0010-perf-gfx-render-pipeline-and-dirty-regions.md","status":"abandoned","created_at":"2026-03-27","updated_at":"2026-06-28","_override_reason":"User explicitly chose to remove this agenda because it no longer applies to the current render scenario; future optimizations can be discussed when needed."}],"decisions":[],"plans":[],"lessons":[],"_override_reason":"User explicitly chose to abandon this discussion because it no longer applies to the current render scenario; future optimizations can be discussed when needed."}
 {"type":"discussion","id":"DSC-0012","status":"done","ticket":"perf-runtime-introspection-syscalls","title":"Agenda - [PERF] Runtime Introspection Syscalls","created_at":"2026-03-27","updated_at":"2026-04-19","tags":["perf","runtime","syscall","telemetry","debug","asset"],"agendas":[],"decisions":[],"plans":[],"lessons":[{"id":"LSN-0034","file":"discussion/lessons/DSC-0012-perf-runtime-introspection-syscalls/LSN-0034-host-owned-debug-boundaries.md","status":"done","created_at":"2026-04-19","updated_at":"2026-04-19"}]}

discussion/workflow/plans/PLN-0128-async-asset-pipeline-contract-tests-and-telemetry.md

@@ -2,7 +2,7 @@
 id: PLN-0128
 ticket: perf-async-background-work-lanes-for-assets-and-fs
 title: Async Asset Pipeline Contract Tests and Telemetry
-status: open
+status: done
 created: 2026-06-28
 completed:
 ref_decisions: [DEC-0034]