fix(kernel): KRNBUG-IO-004 — real XamNotifyCreateListener + XNotifyGetNext per canary

Canary's RegisterNotifyListener (kernel_state.cc:1013-1033) auto-enqueues four
startup notifications on the first listener whose mask covers kXNotifySystem
(SystemUI=0x09 + SystemSignInChanged=0x0A) and kXNotifyLive
(LiveConnectionChanged=0x02000001 + LiveLinkStateChanged=0x02000003). XNotifyGetNext
(xam_notify.cc:22-96) pops the queue with mask + version filtering on enqueue per
xnotifylistener.cc:38-51. Our prior stubs returned 0 forever; the dispatch loop
at 0x822f1be8 in sub_822F1AA8 was thus bypassed indefinitely.

Implementation:
- KernelObject::NotifyListener { mask, max_version, queue, waiters } variant.
- KernelState::has_notified_startup + has_notified_live_startup gates.
- xam_notify_create_listener: mask=r3 (qword), max_version=r4 (clamped <=10),
  alloc handle, conditional 4-tuple startup enqueue.
- xnotify_get_next: handle/match_id/id_ptr/param_ptr in r3..r6; pop_front
  (or scan-by-id), with mask + version filter applied at enqueue time.
- 5 unit tests covering: full-mask 4 startup notifications, second-listener
  no re-fire, system-only mask filtering, max_version=0 too-new drop,
  unknown handle returning 0.

Tests: 594 -> 599. Lockstep `-n 100M` instructions=100000012 deterministic
across 2 reruns; bit-identical run-to-run diff.

Cascade (verified at -n 500M):
- dispatch arm 0x822f1be8 fires; sub_82173DC8 entered.
- 3/21 renderer-cluster L1 PCs newly reached: 0x822c6870 (2 workers),
  0x824563e0, 0x823ddb50.
- canary-only export delta 7 -> 3 (reclassified to fired:
  KeResetEvent, ObCreateSymbolicLink, XamTaskCloseHandle, XamTaskSchedule).
- worker thread count 18 -> 20.
- signal_attempts on handle 0x15e0 = 1 (primary=1), was 0.
- draws=0 still expected at this step.

LOC: 119 (97 impl + 22 scaffolding pattern matches across main.rs / objects.rs
/ state.rs) <= 120.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/xenia-app/src/main.rs

@@ -3230,6 +3230,9 @@ fn dump_thread_diagnostic(
             KernelObject::Mutex { owner, recursion, waiters } => {
                 (format!("Mutex(owner={:?}, rec={})", owner, recursion), waiters)
             }
+            KernelObject::NotifyListener { mask, queue, waiters, .. } => {
+                (format!("NotifyListener(mask={:#x}, pending={})", mask, queue.len()), waiters)
+            }
             KernelObject::File { .. } => continue,
         };
         if waiters.is_empty() {
@@ -3281,7 +3284,8 @@ fn dump_thread_diagnostic(
                     | KernelObject::Semaphore { waiters, .. }
                     | KernelObject::Thread { waiters, .. }
                     | KernelObject::Timer { waiters, .. }
-                    | KernelObject::Mutex { waiters, .. } => Some(waiters.len()),
+                    | KernelObject::Mutex { waiters, .. }
+                    | KernelObject::NotifyListener { waiters, .. } => Some(waiters.len()),
                     KernelObject::File { .. } => None,
                 })
                 .unwrap_or(0);
@@ -3357,7 +3361,8 @@ fn dump_thread_diagnostic(
                         | KernelObject::Semaphore { waiters, .. }
                         | KernelObject::Thread { waiters, .. }
                         | KernelObject::Timer { waiters, .. }
-                        | KernelObject::Mutex { waiters, .. } => Some(waiters.len()),
+                        | KernelObject::Mutex { waiters, .. }
+                        | KernelObject::NotifyListener { waiters, .. } => Some(waiters.len()),
                         KernelObject::File { .. } => None,
                     })
                     .unwrap_or(0);
@@ -3515,7 +3520,8 @@ fn dump_thread_diagnostic(
                         | KernelObject::Semaphore { waiters, .. }
                         | KernelObject::Timer { waiters, .. }
                         | KernelObject::Thread { waiters, .. }
-                        | KernelObject::Mutex { waiters, .. } => waiters.iter().copied().collect(),
+                        | KernelObject::Mutex { waiters, .. }
+                        | KernelObject::NotifyListener { waiters, .. } => waiters.iter().copied().collect(),
                         KernelObject::File { .. } => Vec::new(),
                     };
                     for w in waiters {

crates/xenia-kernel/src/objects.rs

@@ -1,5 +1,6 @@
 //! Kernel object tracking for HLE.
 
+use std::collections::VecDeque;
 use std::sync::Arc;
 
 use xenia_cpu::ThreadRef;
@@ -74,6 +75,12 @@ pub enum KernelObject {
         recursion: u32,
         waiters: Vec<ThreadRef>,
     },
+    NotifyListener {
+        mask: u64,
+        max_version: u32,
+        queue: VecDeque<(u32, u32)>,
+        waiters: Vec<ThreadRef>,
+    },
 }
 
 impl KernelObject {
@@ -87,7 +94,8 @@ impl KernelObject {
             | KernelObject::Semaphore { waiters, .. }
             | KernelObject::Thread { waiters, .. }
             | KernelObject::Timer { waiters, .. }
-            | KernelObject::Mutex { waiters, .. } => Some(waiters),
+            | KernelObject::Mutex { waiters, .. }
+            | KernelObject::NotifyListener { waiters, .. } => Some(waiters),
             KernelObject::File { .. } => None,
         }
     }

crates/xenia-kernel/src/state.rs

@@ -101,6 +101,12 @@ pub struct KernelState {
     /// privilege numbers have already produced a `tracing::info!` line so
     /// the import-hot path doesn't spam at -n 500M.
     pub xex_priv_logged: std::collections::HashSet<u32>,
+    /// Whether the first listener whose mask covers `kXNotifySystem` has
+    /// been registered — gate for the two startup system notifications
+    /// per `kernel_state.cc:1020-1025`.
+    pub has_notified_startup: bool,
+    /// Same for `kXNotifyLive` per `kernel_state.cc:1026-1032`.
+    pub has_notified_live_startup: bool,
     /// Next thread ID. M2.4: atomic.
     pub next_thread_id: std::sync::atomic::AtomicU32,
     /// Virtual file system for NtCreateFile/NtReadFile/etc. The app mounts
@@ -269,6 +275,8 @@ impl KernelState {
             image_base: 0,
             xex_system_flags: 0,
             xex_priv_logged: std::collections::HashSet::new(),
+            has_notified_startup: false,
+            has_notified_live_startup: false,
             next_thread_id: AtomicU32::new(1),
             vfs: None,
             ui: None,

crates/xenia-kernel/src/xam.rs

@@ -355,14 +355,113 @@ fn xam_get_system_version(ctx: &mut PpcContext, _mem: &GuestMemory, _state: &mut
 
 // ===== Notify =====
 
-fn xam_notify_create_listener(ctx: &mut PpcContext, _mem: &GuestMemory, state: &mut KernelState) {
+const K_X_NOTIFY_SYSTEM: u64 = 0x00000001;
-    let handle = state.alloc_handle();
+const K_X_NOTIFY_LIVE: u64 = 0x00000002;
+const K_X_NOTIFICATION_SYSTEM_UI: u32 = 0x09;
+const K_X_NOTIFICATION_SYSTEM_SIGN_IN_CHANGED: u32 = 0x0A;
+const K_X_NOTIFICATION_LIVE_CONNECTION_CHANGED: u32 = 0x02000001;
+const K_X_NOTIFICATION_LIVE_LINK_STATE_CHANGED: u32 = 0x02000003;
+
+fn notification_mask_index(id: u32) -> u32 {
+    (id >> 25) & 0x3F
+}
+
+fn notification_version(id: u32) -> u32 {
+    (id >> 16) & 0x1FF
+}
+
+fn enqueue_notification(obj: &mut KernelObject, id: u32, data: u32) {
+    if let KernelObject::NotifyListener { mask, max_version, queue, .. } = obj {
+        let idx = notification_mask_index(id);
+        if (*mask & (1u64 << idx)) == 0 {
+            return;
+        }
+        if notification_version(id) > *max_version {
+            return;
+        }
+        queue.push_back((id, data));
+    }
+}
+
+fn xam_notify_create_listener(ctx: &mut PpcContext, mem: &GuestMemory, state: &mut KernelState) {
+    let mask = ctx.gpr[3];
+    let mut max_version = ctx.gpr[4] as u32;
+    if max_version > 10 {
+        max_version = 10;
+    }
+    let handle = state.alloc_handle_for(KernelObject::NotifyListener {
+        mask,
+        max_version,
+        queue: std::collections::VecDeque::new(),
+        waiters: Vec::new(),
+    });
+
+    let mut startup_pending: Vec<(u32, u32)> = Vec::new();
+    if !state.has_notified_startup && (mask & K_X_NOTIFY_SYSTEM) != 0 {
+        state.has_notified_startup = true;
+        startup_pending.push((K_X_NOTIFICATION_SYSTEM_UI, 0));
+        startup_pending.push((K_X_NOTIFICATION_SYSTEM_SIGN_IN_CHANGED, 1));
+    }
+    if !state.has_notified_live_startup && (mask & K_X_NOTIFY_LIVE) != 0 {
+        state.has_notified_live_startup = true;
+        startup_pending.push((K_X_NOTIFICATION_LIVE_CONNECTION_CHANGED, 0x001510F1));
+        startup_pending.push((K_X_NOTIFICATION_LIVE_LINK_STATE_CHANGED, 0));
+    }
+    if let Some(obj) = state.objects.get_mut(&handle) {
+        for (id, data) in startup_pending {
+            enqueue_notification(obj, id, data);
+        }
+    }
+
+    let _ = mem;
     ctx.gpr[3] = handle as u64;
 }
 
-fn xnotify_get_next(ctx: &mut PpcContext, _mem: &GuestMemory, _state: &mut KernelState) {
+fn xnotify_get_next(ctx: &mut PpcContext, mem: &GuestMemory, state: &mut KernelState) {
-    // r3 = handle, r4 = id_ptr, r5 = param_ptr
+    let handle = ctx.gpr[3] as u32;
-    ctx.gpr[3] = 0; // FALSE (no notifications)
+    let match_id = ctx.gpr[4] as u32;
+    let id_ptr = ctx.gpr[5] as u32;
+    let param_ptr = ctx.gpr[6] as u32;
+
+    if param_ptr != 0 {
+        mem.write_u32(param_ptr, 0);
+    }
+    if id_ptr == 0 {
+        ctx.gpr[3] = 0;
+        return;
+    }
+    mem.write_u32(id_ptr, 0);
+
+    let Some(KernelObject::NotifyListener { queue, .. }) = state.objects.get_mut(&handle) else {
+        ctx.gpr[3] = 0;
+        return;
+    };
+
+    let dequeued = if match_id != 0 {
+        let pos = queue.iter().position(|&(id, _)| id == match_id);
+        match pos {
+            Some(p) => {
+                let (id, data) = queue.remove(p).unwrap();
+                Some((id, data))
+            }
+            None => None,
+        }
+    } else {
+        queue.pop_front()
+    };
+
+    match dequeued {
+        Some((id, data)) => {
+            mem.write_u32(id_ptr, id);
+            if param_ptr != 0 {
+                mem.write_u32(param_ptr, data);
+            }
+            ctx.gpr[3] = 1;
+        }
+        None => {
+            ctx.gpr[3] = 0;
+        }
+    }
 }
 
 // ===== Session =====
@@ -472,4 +571,99 @@ mod tests {
         xget_avpack(&mut ctx, &mem, &mut state);
         assert_eq!(ctx.gpr[3], 8);
     }
+
+    fn drain_notifications(state: &mut KernelState, mem: &GuestMemory, handle: u32) -> Vec<(u32, u32)> {
+        let id_ptr = SCRATCH_BASE + 0x100;
+        let param_ptr = SCRATCH_BASE + 0x104;
+        let mut out = Vec::new();
+        loop {
+            let mut ctx = PpcContext::default();
+            ctx.gpr[3] = handle as u64;
+            ctx.gpr[4] = 0;
+            ctx.gpr[5] = id_ptr as u64;
+            ctx.gpr[6] = param_ptr as u64;
+            xnotify_get_next(&mut ctx, mem, state);
+            if ctx.gpr[3] == 0 {
+                break;
+            }
+            let id = mem.read_u32(id_ptr);
+            let data = mem.read_u32(param_ptr);
+            out.push((id, data));
+            if out.len() > 16 {
+                panic!("runaway dequeue");
+            }
+        }
+        out
+    }
+
+    fn create_listener(state: &mut KernelState, mem: &GuestMemory, mask: u64, max_version: u32) -> u32 {
+        let mut ctx = PpcContext::default();
+        ctx.gpr[3] = mask;
+        ctx.gpr[4] = max_version as u64;
+        xam_notify_create_listener(&mut ctx, mem, state);
+        ctx.gpr[3] as u32
+    }
+
+    #[test]
+    fn first_listener_with_full_mask_gets_4_startup_notifications() {
+        let (_ctx, mem, mut state) = fresh();
+        let h = create_listener(&mut state, &mem, 0x2F, 10);
+        let drained = drain_notifications(&mut state, &mem, h);
+        assert_eq!(
+            drained,
+            vec![
+                (K_X_NOTIFICATION_SYSTEM_UI, 0),
+                (K_X_NOTIFICATION_SYSTEM_SIGN_IN_CHANGED, 1),
+                (K_X_NOTIFICATION_LIVE_CONNECTION_CHANGED, 0x001510F1),
+                (K_X_NOTIFICATION_LIVE_LINK_STATE_CHANGED, 0),
+            ]
+        );
+    }
+
+    #[test]
+    fn second_listener_does_not_re_fire_startup() {
+        let (_ctx, mem, mut state) = fresh();
+        let h1 = create_listener(&mut state, &mem, 0x2F, 10);
+        let _ = drain_notifications(&mut state, &mem, h1);
+        let h2 = create_listener(&mut state, &mem, 0x2F, 10);
+        assert!(drain_notifications(&mut state, &mem, h2).is_empty());
+    }
+
+    #[test]
+    fn system_only_mask_filters_live() {
+        let (_ctx, mem, mut state) = fresh();
+        let h = create_listener(&mut state, &mem, K_X_NOTIFY_SYSTEM, 10);
+        let drained = drain_notifications(&mut state, &mem, h);
+        assert_eq!(
+            drained,
+            vec![
+                (K_X_NOTIFICATION_SYSTEM_UI, 0),
+                (K_X_NOTIFICATION_SYSTEM_SIGN_IN_CHANGED, 1),
+            ]
+        );
+    }
+
+    #[test]
+    fn version_filter_drops_too_new() {
+        let (_ctx, mem, mut state) = fresh();
+        let h = create_listener(&mut state, &mem, 0x2F, 0);
+        let drained = drain_notifications(&mut state, &mem, h);
+        let kept_versions: Vec<u32> = drained.iter().map(|&(id, _)| notification_version(id)).collect();
+        assert!(kept_versions.iter().all(|&v| v == 0));
+    }
+
+    #[test]
+    fn xnotify_get_next_returns_zero_for_unknown_handle() {
+        let (mut ctx, mem, mut state) = fresh();
+        let id_ptr = SCRATCH_BASE + 0x100;
+        let param_ptr = SCRATCH_BASE + 0x104;
+        ctx.gpr[3] = 0xDEAD_BEEF;
+        ctx.gpr[4] = 0;
+        ctx.gpr[5] = id_ptr as u64;
+        ctx.gpr[6] = param_ptr as u64;
+        xnotify_get_next(&mut ctx, &mem, &mut state);
+        assert_eq!(ctx.gpr[3], 0);
+        assert_eq!(mem.read_u32(id_ptr), 0);
+        assert_eq!(mem.read_u32(param_ptr), 0);
+    }
 }