implements PLN-0128 asset pipeline telemetry tests

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

@@ -112,6 +112,12 @@ pub enum LoadStatus {
     CANCELED = 4,
     ERROR = 5,
     UnknownHandle = 6,
+    QUEUED = 7,
+    ACTIVE = 8,
+    SUPERSEDED = 9,
+    EMPTY = 10,
+    INVALID = 11,
+    BackendUnavailable = 12,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -129,6 +135,36 @@ pub enum AssetOpStatus {
     Ok = 0,
     UnknownHandle = 1,
     InvalidState = 2,
+    Superseded = 3,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct AssetBacklogInfo {
+    pub status: AssetOpStatus,
+    pub pending_count: u32,
+    pub active_handle: HandleId,
+    pub active_asset_id: Option<AssetId>,
+    pub active_bank_type: Option<BankType>,
+    pub active_slot: Option<usize>,
+    pub active_progress: u16,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct AssetBacklogPosition {
+    pub status: AssetOpStatus,
+    pub state: LoadStatus,
+    pub position: u32,
+    pub progress: u16,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct AssetTargetStatus {
+    pub status: AssetOpStatus,
+    pub asset_id: Option<AssetId>,
+    pub handle: HandleId,
+    pub state: LoadStatus,
+    pub position: u32,
+    pub progress: u16,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

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

@@ -1,6 +1,7 @@
 use crate::asset::{
-    AssetEntry, AssetId, AssetLoadError, AssetOpStatus, BankTelemetry, HandleId, LoadStatus,
-    PreloadEntry, SlotRef, SlotStats,
+    AssetBacklogInfo, AssetBacklogPosition, AssetEntry, AssetId, AssetLoadError, AssetOpStatus,
+    AssetTargetStatus, BankTelemetry, BankType, HandleId, LoadStatus, PreloadEntry, SlotRef,
+    SlotStats,
 };
 use crate::cartridge::AssetsPayloadSource;
 
@@ -15,6 +16,12 @@ pub trait AssetBridge {
     fn status(&self, handle: HandleId) -> LoadStatus;
     fn commit(&self, handle: HandleId) -> AssetOpStatus;
     fn cancel(&self, handle: HandleId) -> AssetOpStatus;
+    fn backlog_info(&self) -> AssetBacklogInfo;
+    fn backlog_position(&self, handle: HandleId) -> AssetBacklogPosition;
+    fn backlog_move(&self, handle: HandleId, new_position: usize) -> AssetOpStatus;
+    fn backlog_promote(&self, handle: HandleId) -> AssetOpStatus;
+    fn backlog_demote(&self, handle: HandleId) -> AssetOpStatus;
+    fn target_status(&self, bank_type: BankType, slot: usize) -> AssetTargetStatus;
     fn apply_commits(&self);
     fn bank_telemetry(&self) -> Vec<BankTelemetry>;
     fn slot_info(&self, slot: SlotRef) -> SlotStats;

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

@@ -69,6 +69,12 @@ pub enum Syscall {
     AssetStatus = 0x6002,
     AssetCommit = 0x6003,
     AssetCancel = 0x6004,
+    AssetBacklogInfo = 0x6005,
+    AssetBacklogPosition = 0x6006,
+    AssetBacklogMove = 0x6007,
+    AssetBacklogPromote = 0x6008,
+    AssetBacklogDemote = 0x6009,
+    AssetTargetStatus = 0x600A,
     BankInfo = 0x6101,
 }
 

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

@@ -24,4 +24,40 @@ pub(crate) const ENTRIES: &[SyscallRegistryEntry] = &[
         .caps(caps::ASSET)
         .non_deterministic()
         .cost(20),
+    SyscallRegistryEntry::builder(Syscall::AssetBacklogInfo, "asset", "backlog_info")
+        .args(0)
+        .rets(7)
+        .caps(caps::ASSET)
+        .non_deterministic()
+        .cost(5),
+    SyscallRegistryEntry::builder(Syscall::AssetBacklogPosition, "asset", "backlog_position")
+        .args(1)
+        .rets(4)
+        .caps(caps::ASSET)
+        .non_deterministic()
+        .cost(5),
+    SyscallRegistryEntry::builder(Syscall::AssetBacklogMove, "asset", "backlog_move")
+        .args(2)
+        .rets(1)
+        .caps(caps::ASSET)
+        .non_deterministic()
+        .cost(10),
+    SyscallRegistryEntry::builder(Syscall::AssetBacklogPromote, "asset", "backlog_promote")
+        .args(1)
+        .rets(1)
+        .caps(caps::ASSET)
+        .non_deterministic()
+        .cost(10),
+    SyscallRegistryEntry::builder(Syscall::AssetBacklogDemote, "asset", "backlog_demote")
+        .args(1)
+        .rets(1)
+        .caps(caps::ASSET)
+        .non_deterministic()
+        .cost(10),
+    SyscallRegistryEntry::builder(Syscall::AssetTargetStatus, "asset", "target_status")
+        .args(2)
+        .rets(6)
+        .caps(caps::ASSET)
+        .non_deterministic()
+        .cost(5),
 ];

crates/console/prometeu-hal/src/syscalls/registry.rs

@@ -53,6 +53,12 @@ impl Syscall {
             0x6002 => Some(Self::AssetStatus),
             0x6003 => Some(Self::AssetCommit),
             0x6004 => Some(Self::AssetCancel),
+            0x6005 => Some(Self::AssetBacklogInfo),
+            0x6006 => Some(Self::AssetBacklogPosition),
+            0x6007 => Some(Self::AssetBacklogMove),
+            0x6008 => Some(Self::AssetBacklogPromote),
+            0x6009 => Some(Self::AssetBacklogDemote),
+            0x600A => Some(Self::AssetTargetStatus),
             0x6101 => Some(Self::BankInfo),
             _ => None,
         }
@@ -109,6 +115,12 @@ impl Syscall {
             Self::AssetStatus => "AssetStatus",
             Self::AssetCommit => "AssetCommit",
             Self::AssetCancel => "AssetCancel",
+            Self::AssetBacklogInfo => "AssetBacklogInfo",
+            Self::AssetBacklogPosition => "AssetBacklogPosition",
+            Self::AssetBacklogMove => "AssetBacklogMove",
+            Self::AssetBacklogPromote => "AssetBacklogPromote",
+            Self::AssetBacklogDemote => "AssetBacklogDemote",
+            Self::AssetTargetStatus => "AssetTargetStatus",
             Self::BankInfo => "BankInfo",
         }
     }

crates/console/prometeu-hal/src/syscalls/tests.rs

@@ -258,6 +258,30 @@ fn status_first_syscall_signatures_are_pinned() {
     assert_eq!(asset_cancel.arg_slots, 1);
     assert_eq!(asset_cancel.ret_slots, 1);
 
+    let asset_backlog_info = meta_for(Syscall::AssetBacklogInfo);
+    assert_eq!(asset_backlog_info.arg_slots, 0);
+    assert_eq!(asset_backlog_info.ret_slots, 7);
+
+    let asset_backlog_position = meta_for(Syscall::AssetBacklogPosition);
+    assert_eq!(asset_backlog_position.arg_slots, 1);
+    assert_eq!(asset_backlog_position.ret_slots, 4);
+
+    let asset_backlog_move = meta_for(Syscall::AssetBacklogMove);
+    assert_eq!(asset_backlog_move.arg_slots, 2);
+    assert_eq!(asset_backlog_move.ret_slots, 1);
+
+    let asset_backlog_promote = meta_for(Syscall::AssetBacklogPromote);
+    assert_eq!(asset_backlog_promote.arg_slots, 1);
+    assert_eq!(asset_backlog_promote.ret_slots, 1);
+
+    let asset_backlog_demote = meta_for(Syscall::AssetBacklogDemote);
+    assert_eq!(asset_backlog_demote.arg_slots, 1);
+    assert_eq!(asset_backlog_demote.ret_slots, 1);
+
+    let asset_target_status = meta_for(Syscall::AssetTargetStatus);
+    assert_eq!(asset_target_status.arg_slots, 2);
+    assert_eq!(asset_target_status.ret_slots, 6);
+
     let bank_info = meta_for(Syscall::BankInfo);
     assert_eq!(bank_info.arg_slots, 1);
     assert_eq!(bank_info.ret_slots, 2);

crates/console/prometeu-system/src/lib.rs

@@ -7,7 +7,13 @@ pub use crash_report::CrashReport;
 pub use os::{LifecycleError, LifecycleOperation, SystemOS};
 pub use programs::{NativeShellApp, PrometeuHub, SystemProfileAction, SystemProfileUpdate};
 pub use prometeu_hal::{RenderWorkerBackend, RenderWorkerFrameSink};
+pub use services::async_work::{
+    AsyncWorkActiveJob, AsyncWorkCancelToken, AsyncWorkJobContext, AsyncWorkJobId,
+    AsyncWorkJobKind, AsyncWorkJobOutcome, AsyncWorkLane, AsyncWorkLaneConfig,
+    AsyncWorkLaneController, AsyncWorkLaneError, AsyncWorkLaneTelemetry, AsyncWorkPriority,
+};
 pub use services::fs;
+pub use services::memcard::MemcardAsyncLaneOperation;
 pub use services::process;
 pub use services::task;
 pub use services::vm_runtime::{

crates/console/prometeu-system/src/services/async_work.rs

@@ -0,0 +1,599 @@
+use std::collections::VecDeque;
+use std::panic::{AssertUnwindSafe, catch_unwind};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::mpsc::{self, Receiver, RecvTimeoutError};
+use std::sync::{Arc, Condvar, Mutex};
+use std::thread::{self, JoinHandle};
+use std::time::Duration;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct AsyncWorkJobId(u64);
+
+impl AsyncWorkJobId {
+    pub const ZERO: Self = Self(0);
+
+    pub fn new(value: u64) -> Self {
+        Self(value)
+    }
+
+    pub fn get(self) -> u64 {
+        self.0
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AsyncWorkJobKind {
+    Asset,
+    Memcard,
+    Fs,
+    Test,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AsyncWorkPriority {
+    MemcardCommitWrite,
+    FsWriteConfig,
+    AssetLoad,
+    NonCriticalReadList,
+}
+
+impl AsyncWorkPriority {
+    const fn rank(self) -> u8 {
+        match self {
+            Self::MemcardCommitWrite => 0,
+            Self::FsWriteConfig => 1,
+            Self::AssetLoad => 2,
+            Self::NonCriticalReadList => 3,
+        }
+    }
+}
+
+impl AsyncWorkJobKind {
+    pub const fn default_priority(self) -> AsyncWorkPriority {
+        match self {
+            Self::Memcard => AsyncWorkPriority::MemcardCommitWrite,
+            Self::Fs => AsyncWorkPriority::FsWriteConfig,
+            Self::Asset => AsyncWorkPriority::AssetLoad,
+            Self::Test => AsyncWorkPriority::NonCriticalReadList,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AsyncWorkJobOutcome {
+    Completed,
+    Canceled,
+    Failed,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AsyncWorkLaneError {
+    Shutdown,
+    WorkerPanic,
+    ShutdownTimeout,
+    InternalFailure,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct AsyncWorkLaneConfig {
+    pub shutdown_timeout: Duration,
+}
+
+impl Default for AsyncWorkLaneConfig {
+    fn default() -> Self {
+        Self { shutdown_timeout: Duration::from_millis(250) }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct AsyncWorkActiveJob {
+    pub id: AsyncWorkJobId,
+    pub kind: AsyncWorkJobKind,
+    pub priority: AsyncWorkPriority,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub struct AsyncWorkLaneTelemetry {
+    pub submitted_jobs: u64,
+    pub started_jobs: u64,
+    pub completed_jobs: u64,
+    pub canceled_jobs: u64,
+    pub failed_jobs: u64,
+    pub shutdown_discards: u64,
+    pub pending_depth: usize,
+    pub max_pending_depth: usize,
+    pub active_job_id: Option<AsyncWorkJobId>,
+    pub last_submitted_job_id: Option<AsyncWorkJobId>,
+    pub last_started_job_id: Option<AsyncWorkJobId>,
+    pub last_closed_job_id: Option<AsyncWorkJobId>,
+}
+
+impl AsyncWorkLaneTelemetry {
+    fn record_submitted(&mut self, job_id: AsyncWorkJobId, pending_depth: usize) {
+        self.submitted_jobs += 1;
+        self.pending_depth = pending_depth;
+        self.max_pending_depth = self.max_pending_depth.max(pending_depth);
+        self.last_submitted_job_id = Some(job_id);
+    }
+
+    fn record_started(&mut self, job_id: AsyncWorkJobId, pending_depth: usize) {
+        self.started_jobs += 1;
+        self.pending_depth = pending_depth;
+        self.active_job_id = Some(job_id);
+        self.last_started_job_id = Some(job_id);
+    }
+
+    fn record_closed(&mut self, job_id: AsyncWorkJobId, outcome: AsyncWorkJobOutcome) {
+        match outcome {
+            AsyncWorkJobOutcome::Completed => self.completed_jobs += 1,
+            AsyncWorkJobOutcome::Canceled => self.canceled_jobs += 1,
+            AsyncWorkJobOutcome::Failed => self.failed_jobs += 1,
+        }
+        self.active_job_id = None;
+        self.last_closed_job_id = Some(job_id);
+    }
+
+    fn record_shutdown_discards(&mut self, count: usize) {
+        self.shutdown_discards += count as u64;
+        self.pending_depth = 0;
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct AsyncWorkCancelToken {
+    canceled: Arc<AtomicBool>,
+}
+
+impl AsyncWorkCancelToken {
+    fn new() -> Self {
+        Self { canceled: Arc::new(AtomicBool::new(false)) }
+    }
+
+    pub fn cancel(&self) {
+        self.canceled.store(true, Ordering::Release);
+    }
+
+    pub fn is_canceled(&self) -> bool {
+        self.canceled.load(Ordering::Acquire)
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct AsyncWorkJobContext {
+    pub job_id: AsyncWorkJobId,
+    pub kind: AsyncWorkJobKind,
+    pub priority: AsyncWorkPriority,
+    cancel_token: AsyncWorkCancelToken,
+}
+
+impl AsyncWorkJobContext {
+    pub fn is_canceled(&self) -> bool {
+        self.cancel_token.is_canceled()
+    }
+}
+
+type AsyncWorkFn = Box<dyn FnOnce(AsyncWorkJobContext) -> AsyncWorkJobOutcome + Send + 'static>;
+
+struct PendingAsyncWorkJob {
+    id: AsyncWorkJobId,
+    kind: AsyncWorkJobKind,
+    priority: AsyncWorkPriority,
+    cancel_token: AsyncWorkCancelToken,
+    work: AsyncWorkFn,
+}
+
+struct RunningAsyncWorkJob {
+    id: AsyncWorkJobId,
+    kind: AsyncWorkJobKind,
+    priority: AsyncWorkPriority,
+    cancel_token: AsyncWorkCancelToken,
+    work: AsyncWorkFn,
+}
+
+#[derive(Default)]
+struct AsyncWorkLaneState {
+    next_job_id: u64,
+    pending: VecDeque<PendingAsyncWorkJob>,
+    active: Option<AsyncWorkActiveJob>,
+    active_cancel_token: Option<AsyncWorkCancelToken>,
+    shutdown_requested: bool,
+    telemetry: AsyncWorkLaneTelemetry,
+}
+
+#[derive(Default)]
+pub struct AsyncWorkLane {
+    state: Mutex<AsyncWorkLaneState>,
+    ready: Condvar,
+}
+
+impl AsyncWorkLane {
+    pub fn submit(
+        &self,
+        kind: AsyncWorkJobKind,
+        work: impl FnOnce(AsyncWorkJobContext) -> AsyncWorkJobOutcome + Send + 'static,
+    ) -> Result<AsyncWorkJobId, AsyncWorkLaneError> {
+        self.submit_with_priority(kind, kind.default_priority(), work)
+    }
+
+    pub fn submit_with_priority(
+        &self,
+        kind: AsyncWorkJobKind,
+        priority: AsyncWorkPriority,
+        work: impl FnOnce(AsyncWorkJobContext) -> AsyncWorkJobOutcome + Send + 'static,
+    ) -> Result<AsyncWorkJobId, AsyncWorkLaneError> {
+        let mut state = self.state.lock().map_err(|_| AsyncWorkLaneError::InternalFailure)?;
+        if state.shutdown_requested {
+            return Err(AsyncWorkLaneError::Shutdown);
+        }
+
+        state.next_job_id += 1;
+        let id = AsyncWorkJobId::new(state.next_job_id);
+        let cancel_token = AsyncWorkCancelToken::new();
+        let job = PendingAsyncWorkJob { id, kind, priority, cancel_token, work: Box::new(work) };
+        let insert_at = state
+            .pending
+            .iter()
+            .position(|pending| priority.rank() < pending.priority.rank())
+            .unwrap_or(state.pending.len());
+        state.pending.insert(insert_at, job);
+        let pending_depth = state.pending.len();
+        state.telemetry.record_submitted(id, pending_depth);
+        self.ready.notify_one();
+        Ok(id)
+    }
+
+    pub fn submit_memcard_commit_write(
+        &self,
+        work: impl FnOnce(AsyncWorkJobContext) -> AsyncWorkJobOutcome + Send + 'static,
+    ) -> Result<AsyncWorkJobId, AsyncWorkLaneError> {
+        self.submit_with_priority(
+            AsyncWorkJobKind::Memcard,
+            AsyncWorkPriority::MemcardCommitWrite,
+            work,
+        )
+    }
+
+    pub fn submit_fs_write_config(
+        &self,
+        work: impl FnOnce(AsyncWorkJobContext) -> AsyncWorkJobOutcome + Send + 'static,
+    ) -> Result<AsyncWorkJobId, AsyncWorkLaneError> {
+        self.submit_with_priority(AsyncWorkJobKind::Fs, AsyncWorkPriority::FsWriteConfig, work)
+    }
+
+    pub fn submit_asset_load(
+        &self,
+        work: impl FnOnce(AsyncWorkJobContext) -> AsyncWorkJobOutcome + Send + 'static,
+    ) -> Result<AsyncWorkJobId, AsyncWorkLaneError> {
+        self.submit_with_priority(AsyncWorkJobKind::Asset, AsyncWorkPriority::AssetLoad, work)
+    }
+
+    pub fn submit_non_critical_read_list(
+        &self,
+        kind: AsyncWorkJobKind,
+        work: impl FnOnce(AsyncWorkJobContext) -> AsyncWorkJobOutcome + Send + 'static,
+    ) -> Result<AsyncWorkJobId, AsyncWorkLaneError> {
+        self.submit_with_priority(kind, AsyncWorkPriority::NonCriticalReadList, work)
+    }
+
+    pub fn request_cancel(&self, job_id: AsyncWorkJobId) -> bool {
+        let state = self.state.lock().unwrap();
+        if state.active.is_some_and(|active| active.id == job_id) {
+            if let Some(token) = &state.active_cancel_token {
+                token.cancel();
+                return true;
+            }
+        }
+
+        for pending in &state.pending {
+            if pending.id == job_id {
+                pending.cancel_token.cancel();
+                return true;
+            }
+        }
+
+        false
+    }
+
+    pub fn active_job(&self) -> Option<AsyncWorkActiveJob> {
+        self.state.lock().unwrap().active
+    }
+
+    pub fn pending_depth(&self) -> usize {
+        self.state.lock().unwrap().pending.len()
+    }
+
+    pub fn telemetry(&self) -> AsyncWorkLaneTelemetry {
+        self.state.lock().unwrap().telemetry
+    }
+
+    pub fn request_shutdown(&self) {
+        let mut state = self.state.lock().unwrap();
+        state.shutdown_requested = true;
+        if let Some(token) = &state.active_cancel_token {
+            token.cancel();
+        }
+        self.ready.notify_all();
+    }
+
+    fn wait_take(&self) -> Option<RunningAsyncWorkJob> {
+        self.wait_take_with_hook(|| {})
+    }
+
+    fn wait_take_with_hook(&self, mut before_wait: impl FnMut()) -> Option<RunningAsyncWorkJob> {
+        let mut state = self.state.lock().unwrap();
+        loop {
+            if state.shutdown_requested {
+                let discarded = state.pending.len();
+                state.pending.clear();
+                state.telemetry.record_shutdown_discards(discarded);
+                return None;
+            }
+
+            if let Some(pending) = state.pending.pop_front() {
+                let active = AsyncWorkActiveJob {
+                    id: pending.id,
+                    kind: pending.kind,
+                    priority: pending.priority,
+                };
+                state.active = Some(active);
+                state.active_cancel_token = Some(pending.cancel_token.clone());
+                let pending_depth = state.pending.len();
+                state.telemetry.record_started(pending.id, pending_depth);
+                return Some(RunningAsyncWorkJob {
+                    id: pending.id,
+                    kind: pending.kind,
+                    priority: pending.priority,
+                    cancel_token: pending.cancel_token,
+                    work: pending.work,
+                });
+            }
+
+            before_wait();
+            state = self.ready.wait(state).unwrap();
+        }
+    }
+
+    fn record_finished(&self, job_id: AsyncWorkJobId, outcome: AsyncWorkJobOutcome) {
+        let mut state = self.state.lock().unwrap();
+        if state.active.is_some_and(|active| active.id == job_id) {
+            state.active = None;
+            state.active_cancel_token = None;
+        }
+        state.telemetry.record_closed(job_id, outcome);
+    }
+}
+
+pub struct AsyncWorkLaneController {
+    config: AsyncWorkLaneConfig,
+    lane: Arc<AsyncWorkLane>,
+    handle: Option<JoinHandle<()>>,
+    done_rx: Receiver<Result<(), AsyncWorkLaneError>>,
+}
+
+impl AsyncWorkLaneController {
+    pub fn start(config: AsyncWorkLaneConfig, lane: Arc<AsyncWorkLane>) -> Self {
+        let worker_lane = Arc::clone(&lane);
+        let (done_tx, done_rx) = mpsc::channel();
+        let handle = thread::spawn(move || {
+            let result = catch_unwind(AssertUnwindSafe(|| run_async_work_loop(worker_lane)))
+                .map_err(|_| AsyncWorkLaneError::WorkerPanic);
+            let _ = done_tx.send(result);
+        });
+
+        Self { config, lane, handle: Some(handle), done_rx }
+    }
+
+    pub fn lane(&self) -> Arc<AsyncWorkLane> {
+        Arc::clone(&self.lane)
+    }
+
+    pub fn stop(&mut self) -> Result<(), AsyncWorkLaneError> {
+        self.lane.request_shutdown();
+        match self.done_rx.recv_timeout(self.config.shutdown_timeout) {
+            Ok(result) => {
+                if let Some(handle) = self.handle.take()
+                    && handle.join().is_err()
+                {
+                    return Err(AsyncWorkLaneError::WorkerPanic);
+                }
+                result
+            }
+            Err(RecvTimeoutError::Timeout) => Err(AsyncWorkLaneError::ShutdownTimeout),
+            Err(RecvTimeoutError::Disconnected) => Err(AsyncWorkLaneError::InternalFailure),
+        }
+    }
+}
+
+impl Drop for AsyncWorkLaneController {
+    fn drop(&mut self) {
+        if self.handle.is_some() {
+            let _ = self.stop();
+        }
+    }
+}
+
+fn run_async_work_loop(lane: Arc<AsyncWorkLane>) {
+    while let Some(job) = lane.wait_take() {
+        let context = AsyncWorkJobContext {
+            job_id: job.id,
+            kind: job.kind,
+            priority: job.priority,
+            cancel_token: job.cancel_token,
+        };
+        let result = catch_unwind(AssertUnwindSafe(|| (job.work)(context)))
+            .unwrap_or(AsyncWorkJobOutcome::Failed);
+        lane.record_finished(job.id, result);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::mpsc;
+
+    #[test]
+    fn async_work_lane_runs_jobs_in_submission_order() {
+        let lane = Arc::new(AsyncWorkLane::default());
+        let mut controller =
+            AsyncWorkLaneController::start(AsyncWorkLaneConfig::default(), Arc::clone(&lane));
+        let (tx, rx) = mpsc::channel();
+
+        lane.submit(AsyncWorkJobKind::Test, {
+            let tx = tx.clone();
+            move |_| {
+                tx.send(1).unwrap();
+                AsyncWorkJobOutcome::Completed
+            }
+        })
+        .unwrap();
+        lane.submit(AsyncWorkJobKind::Test, {
+            let tx = tx.clone();
+            move |_| {
+                tx.send(2).unwrap();
+                AsyncWorkJobOutcome::Completed
+            }
+        })
+        .unwrap();
+
+        assert_eq!(rx.recv().unwrap(), 1);
+        assert_eq!(rx.recv().unwrap(), 2);
+        controller.stop().unwrap();
+
+        let telemetry = lane.telemetry();
+        assert_eq!(telemetry.submitted_jobs, 2);
+        assert_eq!(telemetry.started_jobs, 2);
+        assert_eq!(telemetry.completed_jobs, 2);
+    }
+
+    #[test]
+    fn async_work_lane_runs_higher_priority_io_before_assets() {
+        let lane = Arc::new(AsyncWorkLane::default());
+        let (tx, rx) = mpsc::channel();
+
+        lane.submit_asset_load({
+            let tx = tx.clone();
+            move |ctx| {
+                tx.send((ctx.kind, ctx.priority)).unwrap();
+                AsyncWorkJobOutcome::Completed
+            }
+        })
+        .unwrap();
+        lane.submit_non_critical_read_list(AsyncWorkJobKind::Fs, {
+            let tx = tx.clone();
+            move |ctx| {
+                tx.send((ctx.kind, ctx.priority)).unwrap();
+                AsyncWorkJobOutcome::Completed
+            }
+        })
+        .unwrap();
+        lane.submit_memcard_commit_write({
+            let tx = tx.clone();
+            move |ctx| {
+                tx.send((ctx.kind, ctx.priority)).unwrap();
+                AsyncWorkJobOutcome::Completed
+            }
+        })
+        .unwrap();
+        lane.submit_fs_write_config({
+            let tx = tx.clone();
+            move |ctx| {
+                tx.send((ctx.kind, ctx.priority)).unwrap();
+                AsyncWorkJobOutcome::Completed
+            }
+        })
+        .unwrap();
+
+        let mut controller =
+            AsyncWorkLaneController::start(AsyncWorkLaneConfig::default(), Arc::clone(&lane));
+
+        assert_eq!(
+            rx.recv().unwrap(),
+            (AsyncWorkJobKind::Memcard, AsyncWorkPriority::MemcardCommitWrite)
+        );
+        assert_eq!(rx.recv().unwrap(), (AsyncWorkJobKind::Fs, AsyncWorkPriority::FsWriteConfig));
+        assert_eq!(rx.recv().unwrap(), (AsyncWorkJobKind::Asset, AsyncWorkPriority::AssetLoad));
+        assert_eq!(
+            rx.recv().unwrap(),
+            (AsyncWorkJobKind::Fs, AsyncWorkPriority::NonCriticalReadList)
+        );
+        controller.stop().unwrap();
+    }
+
+    #[test]
+    fn async_work_lane_has_one_active_job() {
+        let lane = Arc::new(AsyncWorkLane::default());
+        let mut controller =
+            AsyncWorkLaneController::start(AsyncWorkLaneConfig::default(), Arc::clone(&lane));
+        let (first_started_tx, first_started_rx) = mpsc::channel();
+        let (release_first_tx, release_first_rx) = mpsc::channel();
+        let (second_started_tx, second_started_rx) = mpsc::channel();
+
+        let first = lane
+            .submit(AsyncWorkJobKind::Test, move |_| {
+                first_started_tx.send(()).unwrap();
+                release_first_rx.recv().unwrap();
+                AsyncWorkJobOutcome::Completed
+            })
+            .unwrap();
+        let second = lane
+            .submit(AsyncWorkJobKind::Test, move |_| {
+                second_started_tx.send(()).unwrap();
+                AsyncWorkJobOutcome::Completed
+            })
+            .unwrap();
+
+        first_started_rx.recv().unwrap();
+        assert_eq!(lane.active_job().unwrap().id, first);
+        assert_eq!(lane.pending_depth(), 1);
+        assert!(second_started_rx.try_recv().is_err());
+
+        release_first_tx.send(()).unwrap();
+        second_started_rx.recv().unwrap();
+        assert_eq!(lane.telemetry().last_started_job_id, Some(second));
+        controller.stop().unwrap();
+    }
+
+    #[test]
+    fn async_work_lane_supports_cooperative_cancellation() {
+        let lane = Arc::new(AsyncWorkLane::default());
+        let mut controller =
+            AsyncWorkLaneController::start(AsyncWorkLaneConfig::default(), Arc::clone(&lane));
+        let (started_tx, started_rx) = mpsc::channel();
+
+        let job_id = lane
+            .submit(AsyncWorkJobKind::Test, move |ctx| {
+                started_tx.send(()).unwrap();
+                while !ctx.is_canceled() {
+                    std::thread::yield_now();
+                }
+                AsyncWorkJobOutcome::Canceled
+            })
+            .unwrap();
+
+        started_rx.recv().unwrap();
+        assert!(lane.request_cancel(job_id));
+        controller.stop().unwrap();
+        assert_eq!(lane.telemetry().canceled_jobs, 1);
+    }
+
+    #[test]
+    fn async_work_lane_rejects_submit_after_shutdown() {
+        let lane = Arc::new(AsyncWorkLane::default());
+        lane.request_shutdown();
+
+        let result = lane.submit(AsyncWorkJobKind::Test, |_| AsyncWorkJobOutcome::Completed);
+        assert_eq!(result, Err(AsyncWorkLaneError::Shutdown));
+    }
+
+    #[test]
+    fn async_work_lane_shutdown_discards_pending_jobs() {
+        let lane = Arc::new(AsyncWorkLane::default());
+        lane.submit(AsyncWorkJobKind::Test, |_| AsyncWorkJobOutcome::Completed).unwrap();
+        assert_eq!(lane.pending_depth(), 1);
+
+        lane.request_shutdown();
+        assert!(lane.wait_take().is_none());
+        assert_eq!(lane.pending_depth(), 0);
+        assert_eq!(lane.telemetry().shutdown_discards, 1);
+    }
+}

crates/console/prometeu-system/src/services/fs/mod.rs

@@ -8,4 +8,4 @@ pub use fs_backend::FsBackend;
 pub use fs_entry::FsEntry;
 pub use fs_error::FsError;
 pub use fs_state::FsState;
-pub use virtual_fs::VirtualFS;
+pub use virtual_fs::{FsAsyncLaneOperation, VirtualFS};

crates/console/prometeu-system/src/services/fs/virtual_fs.rs

@@ -1,4 +1,5 @@
 use crate::fs::{FsBackend, FsEntry, FsError};
+use crate::services::async_work::{AsyncWorkJobKind, AsyncWorkPriority};
 
 /// Virtual Filesystem (VFS) interface for Prometeu.
 ///
@@ -13,6 +14,29 @@ pub struct VirtualFS {
     backend: Option<Box<dyn FsBackend>>,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FsAsyncLaneOperation {
+    Write,
+    Delete,
+    Config,
+    Read,
+    ListDir,
+    Exists,
+}
+
+impl FsAsyncLaneOperation {
+    pub const fn job_kind(self) -> AsyncWorkJobKind {
+        AsyncWorkJobKind::Fs
+    }
+
+    pub const fn priority(self) -> AsyncWorkPriority {
+        match self {
+            Self::Write | Self::Delete | Self::Config => AsyncWorkPriority::FsWriteConfig,
+            Self::Read | Self::ListDir | Self::Exists => AsyncWorkPriority::NonCriticalReadList,
+        }
+    }
+}
+
 impl Default for VirtualFS {
     fn default() -> Self {
         Self::new()
@@ -242,4 +266,18 @@ mod tests {
         assert!(matches!(vfs.delete("/"), Err(FsError::PermissionDenied)));
         assert_eq!(calls.delete.load(Ordering::Relaxed), 0);
     }
+
+    #[test]
+    fn fs_async_lane_operations_classify_priority_without_public_api_changes() {
+        assert_eq!(FsAsyncLaneOperation::Write.job_kind(), AsyncWorkJobKind::Fs);
+        assert_eq!(FsAsyncLaneOperation::Write.priority(), AsyncWorkPriority::FsWriteConfig);
+        assert_eq!(FsAsyncLaneOperation::Delete.priority(), AsyncWorkPriority::FsWriteConfig);
+        assert_eq!(FsAsyncLaneOperation::Config.priority(), AsyncWorkPriority::FsWriteConfig);
+        assert_eq!(FsAsyncLaneOperation::Read.priority(), AsyncWorkPriority::NonCriticalReadList);
+        assert_eq!(
+            FsAsyncLaneOperation::ListDir.priority(),
+            AsyncWorkPriority::NonCriticalReadList
+        );
+        assert_eq!(FsAsyncLaneOperation::Exists.priority(), AsyncWorkPriority::NonCriticalReadList);
+    }
 }

crates/console/prometeu-system/src/services/memcard.rs

@@ -1,4 +1,5 @@
 use crate::fs::{FsError, VirtualFS};
+use crate::services::async_work::{AsyncWorkJobKind, AsyncWorkPriority};
 use std::collections::HashMap;
 
 pub const MEMCARD_SLOT_COUNT: usize = 32;
@@ -50,6 +51,30 @@ pub struct MemcardWriteResult {
     pub bytes_written: u32,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum MemcardAsyncLaneOperation {
+    SlotWrite,
+    SlotCommit,
+    SlotClear,
+    SlotRead,
+    SlotStat,
+}
+
+impl MemcardAsyncLaneOperation {
+    pub const fn job_kind(self) -> AsyncWorkJobKind {
+        AsyncWorkJobKind::Memcard
+    }
+
+    pub const fn priority(self) -> AsyncWorkPriority {
+        match self {
+            Self::SlotWrite | Self::SlotCommit | Self::SlotClear => {
+                AsyncWorkPriority::MemcardCommitWrite
+            }
+            Self::SlotRead | Self::SlotStat => AsyncWorkPriority::NonCriticalReadList,
+        }
+    }
+}
+
 #[derive(Debug, Clone)]
 struct SlotImage {
     payload: Vec<u8>,
@@ -428,4 +453,29 @@ mod tests {
         let stat = mem.slot_stat(&fs, 7, 2);
         assert_eq!(stat.state, MemcardSlotState::Corrupt);
     }
+
+    #[test]
+    fn memcard_async_lane_operations_classify_priority_without_changing_public_api() {
+        assert_eq!(MemcardAsyncLaneOperation::SlotWrite.job_kind(), AsyncWorkJobKind::Memcard);
+        assert_eq!(
+            MemcardAsyncLaneOperation::SlotWrite.priority(),
+            AsyncWorkPriority::MemcardCommitWrite
+        );
+        assert_eq!(
+            MemcardAsyncLaneOperation::SlotCommit.priority(),
+            AsyncWorkPriority::MemcardCommitWrite
+        );
+        assert_eq!(
+            MemcardAsyncLaneOperation::SlotClear.priority(),
+            AsyncWorkPriority::MemcardCommitWrite
+        );
+        assert_eq!(
+            MemcardAsyncLaneOperation::SlotRead.priority(),
+            AsyncWorkPriority::NonCriticalReadList
+        );
+        assert_eq!(
+            MemcardAsyncLaneOperation::SlotStat.priority(),
+            AsyncWorkPriority::NonCriticalReadList
+        );
+    }
 }

crates/console/prometeu-system/src/services/mod.rs

@@ -1,3 +1,4 @@
+pub mod async_work;
 pub mod fs;
 pub mod memcard;
 pub mod process;

crates/console/prometeu-system/src/services/vm_runtime/dispatch.rs

@@ -757,6 +757,64 @@ impl NativeInterface for VmRuntimeHost<'_> {
                 ret.push_int(status as i64);
                 Ok(())
             }
+            Syscall::AssetBacklogInfo => {
+                let info = platform.assets().backlog_info();
+                ret.push_int(info.status as i64);
+                ret.push_int(info.pending_count as i64);
+                ret.push_int(info.active_handle as i64);
+                ret.push_int(info.active_asset_id.unwrap_or(0) as i64);
+                ret.push_int(info.active_bank_type.map(|bank| bank as i64).unwrap_or(0));
+                ret.push_int(info.active_slot.map(|slot| slot as i64).unwrap_or(0));
+                ret.push_int(info.active_progress as i64);
+                Ok(())
+            }
+            Syscall::AssetBacklogPosition => {
+                let position = platform.assets().backlog_position(expect_int(args, 0)? as u32);
+                ret.push_int(position.status as i64);
+                ret.push_int(position.state as i64);
+                ret.push_int(position.position as i64);
+                ret.push_int(position.progress as i64);
+                Ok(())
+            }
+            Syscall::AssetBacklogMove => {
+                let new_position = expect_non_negative_usize(args, 1, "new_position")?;
+                let status =
+                    platform.assets().backlog_move(expect_int(args, 0)? as u32, new_position);
+                ret.push_int(status as i64);
+                Ok(())
+            }
+            Syscall::AssetBacklogPromote => {
+                let status = platform.assets().backlog_promote(expect_int(args, 0)? as u32);
+                ret.push_int(status as i64);
+                Ok(())
+            }
+            Syscall::AssetBacklogDemote => {
+                let status = platform.assets().backlog_demote(expect_int(args, 0)? as u32);
+                ret.push_int(status as i64);
+                Ok(())
+            }
+            Syscall::AssetTargetStatus => {
+                let bank_type = match expect_int(args, 0)? {
+                    0 => BankType::GLYPH,
+                    1 => BankType::SOUNDS,
+                    2 => BankType::SCENE,
+                    other => {
+                        return Err(VmFault::Trap(
+                            TRAP_TYPE,
+                            format!("Invalid asset bank type: {}", other),
+                        ));
+                    }
+                };
+                let slot = expect_non_negative_usize(args, 1, "slot")?;
+                let target = platform.assets().target_status(bank_type, slot);
+                ret.push_int(target.status as i64);
+                ret.push_int(target.asset_id.unwrap_or(0) as i64);
+                ret.push_int(target.handle as i64);
+                ret.push_int(target.state as i64);
+                ret.push_int(target.position as i64);
+                ret.push_int(target.progress as i64);
+                Ok(())
+            }
             Syscall::BankInfo => {
                 let asset_type = match expect_int(args, 0)? as u32 {
                     0 => BankType::GLYPH,

crates/console/prometeu-system/src/services/vm_runtime/tests_asset_bank.rs

@@ -180,6 +180,102 @@ fn tick_asset_status_unknown_handle_returns_status_not_crash() {
     assert_eq!(vm.operand_stack_top(1), vec![Value::Int64(LoadStatus::UnknownHandle as i64)]);
 }
 
+#[test]
+fn tick_asset_target_status_empty_slot_returns_status_first_payload() {
+    let mut runtime = VirtualMachineRuntime::new(None);
+    let mut log_service = LogService::new(4096);
+    let mut fs = VirtualFS::new();
+    let mut fs_state = FsState::Unmounted;
+    let mut memcard = MemcardService::new();
+    let mut open_files: HashMap<u32, String> = HashMap::new();
+    let mut next_handle = 1;
+    let mut vm = VirtualMachine::default();
+    let mut platform = TestPlatform::new();
+    let signals = InputSignals::default();
+    let code = assemble("PUSH_I32 0\nPUSH_I32 0\nHOSTCALL 0\nHALT").expect("assemble");
+    let program = serialized_single_function_module(
+        code,
+        vec![SyscallDecl {
+            module: "asset".into(),
+            name: "target_status".into(),
+            version: 1,
+            arg_slots: 2,
+            ret_slots: 6,
+        }],
+    );
+    let cartridge = cartridge_with_program(program, caps::ASSET);
+
+    runtime.initialize_vm(&mut log_service, &mut vm, &cartridge).expect("runtime must initialize");
+    let report = runtime.tick(
+        &mut log_service,
+        &mut fs,
+        &mut fs_state,
+        &mut memcard,
+        &mut open_files,
+        &mut next_handle,
+        &mut vm,
+        &signals,
+        &mut platform,
+    );
+
+    assert!(report.is_none(), "target_status must not crash for empty slots");
+    assert!(vm.is_halted());
+    assert_eq!(
+        vm.operand_stack_top(6),
+        vec![
+            Value::Int64(0),
+            Value::Int64(0),
+            Value::Int64(LoadStatus::EMPTY as i64),
+            Value::Int64(1),
+            Value::Int64(0),
+            Value::Int64(AssetOpStatus::Ok as i64),
+        ]
+    );
+}
+
+#[test]
+fn tick_asset_backlog_promote_unknown_handle_returns_status_not_crash() {
+    let mut runtime = VirtualMachineRuntime::new(None);
+    let mut log_service = LogService::new(4096);
+    let mut fs = VirtualFS::new();
+    let mut fs_state = FsState::Unmounted;
+    let mut memcard = MemcardService::new();
+    let mut open_files: HashMap<u32, String> = HashMap::new();
+    let mut next_handle = 1;
+    let mut vm = VirtualMachine::default();
+    let mut platform = TestPlatform::new();
+    let signals = InputSignals::default();
+    let code = assemble("PUSH_I32 999\nHOSTCALL 0\nHALT").expect("assemble");
+    let program = serialized_single_function_module(
+        code,
+        vec![SyscallDecl {
+            module: "asset".into(),
+            name: "backlog_promote".into(),
+            version: 1,
+            arg_slots: 1,
+            ret_slots: 1,
+        }],
+    );
+    let cartridge = cartridge_with_program(program, caps::ASSET);
+
+    runtime.initialize_vm(&mut log_service, &mut vm, &cartridge).expect("runtime must initialize");
+    let report = runtime.tick(
+        &mut log_service,
+        &mut fs,
+        &mut fs_state,
+        &mut memcard,
+        &mut open_files,
+        &mut next_handle,
+        &mut vm,
+        &signals,
+        &mut platform,
+    );
+
+    assert!(report.is_none(), "unknown backlog handle must not crash");
+    assert!(vm.is_halted());
+    assert_eq!(vm.operand_stack_top(1), vec![Value::Int64(AssetOpStatus::UnknownHandle as i64)]);
+}
+
 #[test]
 fn tick_bank_info_returns_slot_summary_not_json() {
     let mut runtime = VirtualMachineRuntime::new(None);

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":"open","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":"open","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":"open","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":"open","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":"open","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/agendas/AGD-0006-app-home-filesystem-surface-and-semantics.md

@@ -22,6 +22,19 @@ Sem um contrato claro de `home` por app, a API tende a crescer com semantica inc
 1. Todo `app` acessa somente sua `home` logica.
 2. Nunca ha acesso direto ao filesystem global do host pela userland.
 3. O runtime `fs` interno continua cobrindo tanto `game` quanto `app`.
+4. Existe uma async IO lane compartilhavel por assets, memcard e FS. Esta
+   agenda deve decidir a API publica de FS considerando essa lane, mas a
+   existencia da lane ja esta fechada por `DEC-0034`.
+
+## Fronteira com a Async IO Lane
+
+`DEC-0034` fecha a existencia de uma lane serial para trabalho async de IO. FS
+pode consumir essa lane internamente para escrita, configuracao, leitura ou
+listagem, conforme prioridade operacional definida pelo runtime.
+
+Esta agenda continua dona da decisao sobre API publica de FS para `app home`.
+Portanto, a existencia da lane nao implica criar agora request handles, polling
+publico ou novas syscalls de FS. Esses shapes permanecem em aberto aqui.
 
 ## Alvo da Discussao
 
@@ -116,6 +129,8 @@ No perfil `app` (`home` sandbox), esta agenda passa a ser a fonte normativa para
 2. `rename` entra no MVP ou pode ficar para fase seguinte?
 3. Qual conjunto minimo de metadados garante portabilidade real entre hosts?
 4. Qual grau de atomicidade e obrigatorio para escrita de arquivo no v1?
+5. Quais operacoes de FS devem consumir a async IO lane e quais permanecem
+   sincrono-aparentes para a userland?
 
 ## Dependencias
 

discussion/workflow/plans/PLN-0123-async-work-lane-specification-propagation.md

@@ -2,7 +2,7 @@
 id: PLN-0123
 ticket: perf-async-background-work-lanes-for-assets-and-fs
 title: Async Work Lane Specification Propagation
-status: open
+status: done
 created: 2026-06-28
 completed:
 ref_decisions: [DEC-0034]

discussion/workflow/plans/PLN-0124-async-work-lane-runtime-infrastructure.md

@@ -2,7 +2,7 @@
 id: PLN-0124
 ticket: perf-async-background-work-lanes-for-assets-and-fs
 title: Async Work Lane Runtime Infrastructure
-status: open
+status: done
 created: 2026-06-28
 completed:
 ref_decisions: [DEC-0034]

discussion/workflow/plans/PLN-0125-asset-backlog-and-stable-slot-handles.md

@@ -2,7 +2,7 @@
 id: PLN-0125
 ticket: perf-async-background-work-lanes-for-assets-and-fs
 title: Asset Backlog and Stable Slot Handles
-status: open
+status: done
 created: 2026-06-28
 completed:
 ref_decisions: [DEC-0034]

discussion/workflow/plans/PLN-0126-asset-backlog-public-api-and-status-surface.md

@@ -2,7 +2,7 @@
 id: PLN-0126
 ticket: perf-async-background-work-lanes-for-assets-and-fs
 title: Asset Backlog Public API and Status Surface
-status: open
+status: done
 created: 2026-06-28
 completed:
 ref_decisions: [DEC-0034]

discussion/workflow/plans/PLN-0127-async-lane-memcard-and-fs-integration-boundaries.md

@@ -2,7 +2,7 @@
 id: PLN-0127
 ticket: perf-async-background-work-lanes-for-assets-and-fs
 title: Async Lane Memcard and FS Integration Boundaries
-status: open
+status: done
 created: 2026-06-28
 completed:
 ref_decisions: [DEC-0034]

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]

docs/specs/runtime/09-events-and-concurrency.md

@@ -127,7 +127,31 @@ Shutdown is explicit and bounded. A shutdown request wakes a waiting worker,
 causes pending work that will not be consumed to be discarded, and reports a
 typed failure if the worker cannot join within the configured timeout.
 
-## 8 Determinism and Best Practices
+## 8 Async Asset and IO Work Lane
+
+The asset/IO async work lane is not a machine-visible event source and does not
+introduce guest callbacks. It is an implementation-side lane for asset
+IO/decode/materialization and compatible persistence IO work.
+
+The lane is serial:
+
+- it has at most one active job;
+- it keeps an ordered backlog;
+- it is separate from the render worker;
+- it must not create one OS thread per guest-visible asset request.
+
+Asset jobs are keyed by target `bank_type/slot`. A newer request for the same
+target supersedes the older request. Superseding is an operational status and
+does not execute guest code.
+
+Asset install/commit into resident banks happens on the main runtime lane at
+predictable ownership points. The async lane prepares materialized results; it
+does not publish resident graphics/audio/scene state directly.
+
+FS and game persistence services may consume this lane for IO-style work, but
+public FS API shape is defined by the FS/app-home contract, not by this chapter.
+
+## 9 Determinism and Best Practices
 
 PROMETEU encourages:
 
@@ -142,7 +166,7 @@ PROMETEU discourages:
 - hidden timing channels;
 - ambiguous out-of-band execution.
 
-## 9 Relationship to Other Specs
+## 10 Relationship to Other Specs
 
 - [`09a-coroutines-and-cooperative-scheduling.md`](09a-coroutines-and-cooperative-scheduling.md) defines coroutine lifecycle and scheduling behavior.
 - [`10-debug-inspection-and-profiling.md`](10-debug-inspection-and-profiling.md) defines observability and diagnostics surfaces.

docs/specs/runtime/15-asset-management.md

@@ -393,14 +393,129 @@ Fault boundary:
 
 Rules:
 
-- `handle` is valid only when `load` status is `OK`;
+- `handle` is returned when `load` status is `OK`;
+- `handle` represents a stable bank slot target, not a worker thread;
+- a known handle remains queryable even when the slot is empty, has no active
+  request, or has a superseded request;
 - failed `load` returns `handle = 0`;
 - `commit` and `cancel` must not be silent no-op for unknown/invalid handle state.
 - `asset.load` resolves the target bank type from `asset_table` using `asset_id`;
 - public callers must not supply `asset_name` or `bank_type` to `asset.load`;
 - slot validation and residency/lifecycle rejection remain in `asset` status space and are not delegated to `bank`.
 
-### 11.2 Minimum status tables
+### 11.2 Async work lane and backlog
+
+Asset loading uses the runtime async work lane. This lane is separate from the
+VM/main lane and separate from the render worker lane.
+
+The asset async lane is serial:
+
+- it executes at most one active asset job at a time;
+- it keeps an ordered backlog of pending requests;
+- it must not create one OS thread per `asset.load` request.
+
+Asset requests are keyed by the target `bank_type/slot`.
+
+Rules:
+
+- each `bank_type/slot` has at most one current request;
+- a newer request for the same `bank_type/slot` supersedes the previous request;
+- if the previous request is queued, it is removed from the backlog;
+- if the previous request is active, the lane should cancel cooperatively when
+  the current phase supports cheap cancellation;
+- if active work cannot stop cheaply, it may finish, but the result must be
+  discarded when its generation no longer matches the target request generation;
+- if the target already contains the requested `asset_id` as a valid resident
+  asset, `asset.load` returns a ready handle without adding a backlog entry.
+
+The effective backlog size is bounded by the sum of targetable bank slots,
+because only one current request can exist per `bank_type/slot`. The runtime
+does not expose a guest-visible `queue_full` status for the normal asset
+backlog path.
+
+### 11.3 Handle state
+
+An asset handle observes one stable bank slot target. Its observable state has
+two parts:
+
+```text
+handle:
+  bank_type
+  slot
+
+slot_state:
+  loaded_asset_id
+  resident_state
+  slot_generation
+
+request_state:
+  requested_asset_id
+  request_generation
+  state
+  backlog_position
+  progress
+```
+
+`slot_state` describes what is currently resident in the target slot.
+`request_state` describes the current or most recent request for that target.
+
+Mutating operations such as commit, cancel, promote, demote, and move must act
+on the current request generation. They must not accidentally mutate a newer
+request through an older handle view.
+
+### 11.4 Backlog inspection and ordering
+
+The asset backlog surface may expose these status-first operations:
+
+- `asset.backlog_info() -> (status, pending_count, active_handle, active_asset_id, active_bank_type, active_slot, active_progress)`
+- `asset.backlog_position(handle) -> (status, state, position, progress)`
+- `asset.backlog_move(handle, new_position) -> status`
+- `asset.backlog_promote(handle) -> status`
+- `asset.backlog_demote(handle) -> status`
+- `asset.target_status(bank_type, slot) -> (status, asset_id, handle, state, position, progress)`
+
+`asset.backlog_promote(handle)` is a shortcut for moving a queued request to
+position `1`, the first pending position after the active job.
+
+`asset.backlog_demote(handle)` is a shortcut for moving a queued request to the
+end of the pending backlog.
+
+### 11.5 Progress and telemetry
+
+Asset progress uses integer progress, not floating point. The preferred scale is
+`0..10000`.
+
+The initial phase model is:
+
+```text
+queued -> 0
+read   -> 0..4000
+decode -> 4000..9000
+stage  -> 9000..10000
+ready  -> 10000
+```
+
+If a phase cannot report internal progress, it keeps the previous progress mark
+and advances at phase completion. The runtime must not invent false precision
+for non-linear decode phases.
+
+Minimum telemetry:
+
+- current backlog depth;
+- target/request position;
+- active job progress;
+- jobs submitted;
+- jobs completed;
+- jobs failed;
+- jobs canceled;
+- jobs superseded;
+- job duration;
+- percentiles by `bank_type`;
+- lightweight percentiles or small-window samples by `asset_id`.
+
+Percentiles are updated when a job closes, not inside the inner decode loop.
+
+### 11.6 Minimum status tables
 
 `asset.load` request statuses:
 
@@ -418,9 +533,16 @@ Rules:
 - `4` = `CANCELED`
 - `5` = `ERROR`
 - `6` = `UNKNOWN_HANDLE`
+- `7` = `QUEUED`
+- `8` = `ACTIVE`
+- `9` = `SUPERSEDED`
+- `10` = `EMPTY`
+- `11` = `INVALID`
+- `12` = `BACKEND_UNAVAILABLE`
 
 `asset.commit` and `asset.cancel` operation statuses:
 
 - `0` = `OK`
 - `1` = `UNKNOWN_HANDLE`
 - `2` = `INVALID_STATE`
+- `3` = `SUPERSEDED`

docs/specs/runtime/16-host-abi-and-syscalls.md

@@ -203,6 +203,12 @@ Canonical operations in v1 are:
 - `asset.status(handle) -> status`
 - `asset.commit(handle) -> status`
 - `asset.cancel(handle) -> status`
+- `asset.backlog_info() -> (status, pending_count, active_handle, active_asset_id, active_bank_type, active_slot, active_progress)`
+- `asset.backlog_position(handle) -> (status, state, position, progress)`
+- `asset.backlog_move(handle, new_position) -> status`
+- `asset.backlog_promote(handle) -> status`
+- `asset.backlog_demote(handle) -> status`
+- `asset.target_status(bank_type, slot) -> (status, asset_id, handle, state, position, progress)`
 
 For `asset.load`:
 
@@ -210,6 +216,18 @@ For `asset.load`:
 - `slot` is the target slot index;
 - bank kind is resolved from `asset_table` by `asset_id`, not supplied by the caller.
 
+Asset handles represent stable bank slot targets. A handle can be queried even
+when its slot has no resident asset or active request. Internally the handle
+state separates resident `slot_state` from current `request_state`.
+
+The asset backlog is keyed by `bank_type/slot`. New requests for the same target
+supersede older requests for that target. `superseded` is an operational status,
+not a structural trap.
+
+`asset.backlog_promote(handle)` and `asset.backlog_demote(handle)` are
+convenience operations over backlog movement. They do not introduce a second
+ordering model.
+
 ### Bank diagnostics surface (`bank`, v1)
 
 `DEC-0009` narrows the public bank contract:

docs/specs/runtime/16a-syscall-policies.md

@@ -44,6 +44,9 @@ Normal operational success and operational failure conditions should be represen
 Examples:
 
 - asset not yet loaded;
+- asset request queued or active;
+- asset request superseded by a newer request for the same bank slot;
+- asset backend unavailable;
 - audio voice unavailable;
 - persistent storage full.
 
@@ -107,6 +110,11 @@ Game memcard operations (`mem.*`) are status-first and use `fs` capability in v1
 `mem` remains layered on runtime `fs`; no parallel persistence channel is introduced.
 Domain surface, status catalog and slot semantics are defined in [`08-save-memory-and-memcard.md`](08-save-memory-and-memcard.md).
 
+Asset backlog operations are status-first. `queued`, `active`, `ready`,
+`canceled`, `superseded`, `empty`, `invalid`, decode failure, and backend
+unavailability are operational results. They must not be reclassified as `Trap`
+unless the caller violates the structural ABI contract.
+
 ## 3 Interaction with the Garbage Collector
 
 The VM heap and host-managed memory are separate.