diff --git a/drivers/gpu/drm/asahi/g17/context.rs b/drivers/gpu/drm/asahi/g17/context.rs
index 0d3948a2b69ffb..0b3bd97fb42fe4 100644
--- a/drivers/gpu/drm/asahi/g17/context.rs
+++ b/drivers/gpu/drm/asahi/g17/context.rs
@@ -114,6 +114,9 @@ pub(crate) struct Context {
     scheduler: Arc<KernelObject>,
     scheduler_generation: u32,
     published: AtomicBool,
+    /// A kill of this context was published, through a compute queue or on its render pairs'
+    /// behalf. A context is killed at most once.
+    kill_published: AtomicBool,
     /// Advisory only: publication still checks the physical binding under the
     /// device mutex. False skips an inline attempt known to need admission.
     compute_binding_hint: AtomicBool,
@@ -197,6 +200,7 @@ impl Context {
             scheduler,
             scheduler_generation,
             published: AtomicBool::new(false),
+            kill_published: AtomicBool::new(false),
             compute_binding_hint: AtomicBool::new(false),
             render_seen: AtomicBool::new(false),
             qos,
@@ -280,6 +284,28 @@ impl Context {
             _work: self.work.clone(),
         }
     }
+    /// Called once the firmware consumed this context's scheduler release. If its work list
+    /// still links a node, the firmware keeps referencing that node: on Neo a render command
+    /// that ran behind a failed compute command is never unlinked, and a later context given
+    /// the same work storage (the object cache hands it out again) made the firmware recover
+    /// the GPU. Keep the storage mapped and out of reuse until firmware stop; returns whether
+    /// it was linked.
+    pub(crate) fn retain_linked_work(&self, pool: &Arc<Pool>) -> bool {
+        // An unreadable head counts as linked: retaining is the safe side.
+        let linked = self.work_head().map_or(true, |head| head != [0, self.work_head_va()]);
+        if linked {
+            self.work.lock().object.retain_until_stop(pool);
+        }
+        linked
+    }
+
+    /// The firmware may still walk this context's work list into memory it names: a node is
+    /// linked and the scheduler page shows no kill. Unreadable state counts as linked.
+    pub(crate) fn firmware_may_link(&self) -> bool {
+        let killed = super::queue::compute::exit::State::read(self).map_or(0, |state| state.killed);
+        must_retain(self.work_head().ok(), self.work_head_va(), killed)
+    }
+
     pub(crate) fn work_head_va(&self) -> u64 {
         self.work_base + (UAT_PGSZ + 0x400) as u64
     }
@@ -294,6 +320,16 @@ impl Context {
         Ok(head)
     }
 
+    /// Record that a kill of this context was published (see `kill_published`).
+    pub(crate) fn note_kill_published(&self) {
+        self.kill_published.store(true, Ordering::Release);
+    }
+
+    /// A kill of this context was published.
+    pub(crate) fn kill_published(&self) -> bool {
+        self.kill_published.load(Ordering::Acquire)
+    }
+
     pub(crate) fn is_published(&self) -> bool {
         self.published.load(Ordering::Acquire)
     }
@@ -463,10 +499,26 @@ impl QosIds {
     }
 }
 
+/// Memory a context's work list may lead the firmware to stays allocated: a linked node (or
+/// an unreadable head) and no acknowledged kill.
+fn must_retain(head: Option<[u64; 2]>, head_va: u64, killed: u32) -> bool {
+    killed == 0 && head.is_none_or(|head| head != [0, head_va])
+}
+
 #[kunit_tests(asahi_g17_qos_capacity)]
 mod capacity_tests {
     use super::*;
 
+    #[test]
+    fn linked_unkilled_work_is_retained() {
+        let va = 0xfffffc20_0000_4400;
+        assert!(!must_retain(Some([0, va]), va, 0));
+        assert!(must_retain(Some([0xfffffc20_0000_0000, 0xfffffc20_0000_0098]), va, 0));
+        assert!(must_retain(None, va, 0));
+        assert!(!must_retain(Some([0xfffffc20_0000_0000, 0xfffffc20_0000_0098]), va, 1));
+        assert!(!must_retain(None, va, 1));
+    }
+
     #[test]
     fn failed_client_construction_returns_every_qos_id() {
         let ids = QosIds::new().unwrap();
diff --git a/drivers/gpu/drm/asahi/g17/event.rs b/drivers/gpu/drm/asahi/g17/event.rs
index 9508a827be4b1b..a175e583923792 100644
--- a/drivers/gpu/drm/asahi/g17/event.rs
+++ b/drivers/gpu/drm/asahi/g17/event.rs
@@ -36,6 +36,8 @@ pub(super) trait Queues {
     fn retain_render_completions(&mut self, masks: [u64; 2]);
     fn complete_compute(&mut self, masks: [u64; 2]) -> Result;
     fn context_killed(&mut self, cookie: u64);
+    /// The firmware failed a compute command (see [`Event::ComputeFault`]).
+    fn compute_fault(&mut self, qid: u8, stamp: u32);
 }
 
 /// Work requested after releasing the device mutex.
@@ -72,6 +74,12 @@ fn batch<H: Host>(host: &mut H, role: Role, masks: &mut [u64; 2]) -> Result {
             }
             Event::FreeListGrow(completion) => host.complete_freelist_grow(completion)?,
             Event::ContextKilled(cookie) => host.context_killed(cookie),
+            Event::ComputeFault { qid, stamp } => {
+                crate::g17::trace::teardown(1, u64::from(qid), u64::from(stamp), 0);
+                host.compute_fault(qid, stamp);
+                // Its queue retires or settles its quarantine on this pass.
+                masks[usize::from(qid) / 64] |= 1 << (qid % 64);
+            }
             Event::Unknown(_) => {}
         }
         Ok(true)
diff --git a/drivers/gpu/drm/asahi/g17/fw/channels.rs b/drivers/gpu/drm/asahi/g17/fw/channels.rs
index 524132c4ae7beb..53e899085a0e45 100644
--- a/drivers/gpu/drm/asahi/g17/fw/channels.rs
+++ b/drivers/gpu/drm/asahi/g17/fw/channels.rs
@@ -136,6 +136,9 @@ pub(crate) enum Event {
     TvbGrow(TvbGrowRequest),
     FreeListGrow(FreeListCompletion),
     ContextKilled(u64),
+    /// The firmware failed a compute command of queue `qid` and stores no completion stamp for
+    /// it: `stamp` is the value that command would have stored.
+    ComputeFault { qid: u8, stamp: u32 },
     Unknown(u32),
 }
 
@@ -145,6 +148,8 @@ impl EventRecord {
     const TVB_GROW: u32 = 6;
     const FREE_LIST_GROW: u32 = 13;
     const CONTEXT_KILLED: u32 = 12;
+    const COMPUTE_FAULT: u32 = 7;
+    const COMPUTE_QIDS: u32 = 128;
     const DIAGNOSTIC_PAIRS_MAX: u16 = 5;
     const TVB_SLOTS: u32 = 127;
     const SUBPIPES: u32 = 64;
@@ -198,6 +203,19 @@ impl EventRecord {
                     halt_count: record.halt_count,
                 }))
             }
+            Self::COMPUTE_FAULT => {
+                // SAFETY: Every constructor initializes all integer words.
+                let words = unsafe { self.words };
+                // Only compute faults are known to use this type: any other layout stays an
+                // unknown event, consumed and ignored as before.
+                if words[3] >= Self::COMPUTE_QIDS || words[4] & 0xff != 0 {
+                    return Ok(Event::Unknown(Self::COMPUTE_FAULT));
+                }
+                Ok(Event::ComputeFault {
+                    qid: words[3] as u8,
+                    stamp: words[4],
+                })
+            }
             Self::CONTEXT_KILLED => {
                 // SAFETY: Every constructor initializes all integer words.
                 let words = unsafe { self.words };
@@ -525,3 +543,26 @@ impl ContextKill {
         }
     }
 }
+
+#[cfg_attr(not(test), kernel::macros::kunit_tests(asahi_g17_event_records))]
+mod tests {
+    use super::*;
+
+    fn record(words: &[u32]) -> EventRecord {
+        let mut all = [0u32; EVENT_WORDS];
+        all[..words.len()].copy_from_slice(words);
+        EventRecord::from_words(all)
+    }
+
+    #[test]
+    fn a_compute_fault_names_its_queue_and_stamp() {
+        let event = record(&[7, 1, 2, 0x39, 0x300, 0xffff_fc00, 1]).decode();
+        assert_eq!(event, Ok(Event::ComputeFault { qid: 0x39, stamp: 0x300 }));
+    }
+
+    #[test]
+    fn an_unrecognized_type_7_layout_stays_unknown() {
+        assert_eq!(record(&[7, 1, 2, 128, 0x300]).decode(), Ok(Event::Unknown(7)));
+        assert_eq!(record(&[7, 1, 2, 9, 0x301]).decode(), Ok(Event::Unknown(7)));
+    }
+}
diff --git a/drivers/gpu/drm/asahi/g17/fw/render.rs b/drivers/gpu/drm/asahi/g17/fw/render.rs
index aa8693bf7140b7..a6bc1704d1418d 100644
--- a/drivers/gpu/drm/asahi/g17/fw/render.rs
+++ b/drivers/gpu/drm/asahi/g17/fw/render.rs
@@ -1023,7 +1023,11 @@ pub(crate) struct TaDescriptor {
     /// QID of the tiling queue.
     pub(crate) qid: u32,
     pub(crate) unk_8be: u32,
-    pub(crate) unk_8c2: u32,
+    /// Non-zero on the first kick of a queue pair, like the fragment descriptor's word at the
+    /// same position. Set on every kick, the firmware also notifies the host of each pass's
+    /// tiling completion, which no retirement needs. The 25G83 firmware keeps it set on every
+    /// kick until this is measured there.
+    pub(crate) first_kick: u32,
     /// Completion value of the tiling kick (see
     /// [`KickTimestamp::completion_value`]).
     pub(crate) completion: u64,
@@ -1073,7 +1077,6 @@ impl TaDescriptor {
     const ARRAY_OFFSET: u64 = core::mem::offset_of!(Self, registers) as u64;
     const UNK_876: u32 = u32::MAX;
     const UNK_892: u32 = 1;
-    const UNK_8C2: u32 = 1;
     const UNK_932: u8 = 0x44;
     const UNK_93C: u8 = 1;
     const UNK_93E_FIRST_KICK: [u8; 2] = [0xd0, 0x91];
@@ -1127,7 +1130,10 @@ impl TaDescriptor {
         self.aux_stamp_va = args.ta.aux_stamp_va;
         self.stamp_value_2 = stamp_value;
         self.qid = args.ta.qid.into();
-        self.unk_8c2 = Self::UNK_8C2;
+        self.first_kick = match args.programs == RegisterPrograms::V25G83 {
+            true => 1,
+            false => u32::from(args.ordinal == 0),
+        };
         self.current_id = resources.ta_ids.current.into();
         self.ordinal = args.ordinal as u32;
         self.timestamp_start_va = resources.timestamp_va;
diff --git a/drivers/gpu/drm/asahi/g17/job.rs b/drivers/gpu/drm/asahi/g17/job.rs
index 647b20e9753219..29908f214bdc53 100644
--- a/drivers/gpu/drm/asahi/g17/job.rs
+++ b/drivers/gpu/drm/asahi/g17/job.rs
@@ -293,6 +293,8 @@ pub(crate) trait Backend: Send + Sync {
     fn cancel(&self, packet: Arc<Packet>);
     /// Transfer a timeout to an outstanding off-lock graph transaction, if any.
     fn defer_timeout(&self, packet: &Arc<Packet>, renewals: u8) -> bool;
+    /// A long job of another queue is running beside `packet`.
+    fn long_job_beside(&self, _packet: &Arc<Packet>) -> bool { false }
     /// Return true only after taking publication ownership and recording its result.
     fn try_early(&self, packet: &Arc<Packet>) -> bool;
     fn count_backlog(&self, packet: &Packet);
@@ -346,11 +348,11 @@ impl<B: Backend> sched::JobImpl for Job<B> {
         ) {
             return true;
         }
-        // Other work retiring means the GPU is shared, not hung.
-        if super::timeout::renew_shared(
-            job.packet.is_published(), super::recovery::since_retired(super::now_ns()),
-            &mut job.shared_renewals,
-        ) {
+        // Other work retiring means the GPU is shared, not hung. So does another long job
+        // running beside a long job, though neither retires until it ends.
+        let shared = super::timeout::retiring(super::recovery::since_retired(super::now_ns()))
+            || (job.packet.command.long_job() && job.backend.long_job_beside(&job.packet));
+        if super::timeout::renew_shared(job.packet.is_published(), shared, &mut job.shared_renewals) {
             return true;
         }
         let deferred = job.backend.defer_timeout(&job.packet, job.victim_renewals);
diff --git a/drivers/gpu/drm/asahi/g17/kick.rs b/drivers/gpu/drm/asahi/g17/kick.rs
index a1f8f70f597d47..3b55c283254b5e 100644
--- a/drivers/gpu/drm/asahi/g17/kick.rs
+++ b/drivers/gpu/drm/asahi/g17/kick.rs
@@ -43,7 +43,7 @@ const RENDER_HEADROOM: usize = 24;
 /// The render reserve with every execution context open.
 const FULL_RENDER_RESERVE: usize = render_reserve(crate::mmu::MAX_EXECUTION_CONTEXTS as usize);
 /// Compute channels that stay reachable with every execution context open.
-#[cfg(CONFIG_KUNIT)]
+#[cfg(CONFIG_KUNIT = "y")]
 pub(super) const COMPUTE_FLOOR: usize = QID_COUNT - FULL_RENDER_RESERVE;
 /// Upper bound on distinct physical compute channels named by one kick.
 pub(super) const COMPUTE_CHANNELS: usize = QID_COUNT;
@@ -145,6 +145,11 @@ impl QueueIds {
         self.render_ids() + (QID_COUNT - self.floor - self.leased) >= 2 * queues + 2
     }
 
+    /// Whether a render pair not yet built can still reserve both of its identities.
+    pub(crate) fn render_pair_fits(&self) -> bool {
+        QID_COUNT - self.floor - self.leased >= 2
+    }
+
     /// Whether a fresh compute identity fits beside the render reserve of
     /// `queues` open logical queues, without reserving it.
     pub(crate) fn compute_fits(&self, queues: usize) -> bool {
diff --git a/drivers/gpu/drm/asahi/g17/mod.rs b/drivers/gpu/drm/asahi/g17/mod.rs
index a0d23c9d69e1e8..31c242a1323f8d 100644
--- a/drivers/gpu/drm/asahi/g17/mod.rs
+++ b/drivers/gpu/drm/asahi/g17/mod.rs
@@ -65,6 +65,7 @@ use core::ops::Range;
 use core::sync::atomic::{
     AtomicBool,
     AtomicU32,
+    AtomicU64,
     Ordering, //
 };
 
@@ -369,7 +370,7 @@ impl rtkit::Operations for RoleOps {
         if !is_events_message(ep, msg) {
             return false;
         }
-        data.shared.queue_events();
+        data.shared.queue_event_records();
         true
     }
 
@@ -857,6 +858,11 @@ struct Shared {
     compute_demand: AtomicU32,
     #[pin]
     work: Work<Shared, EVENT_WORK>,
+    /// The next event pass runs every housekeeping step: set by every [`Self::queue_events`]
+    /// request, not by the firmware's event notification ([`Self::queue_event_records`]).
+    housekeeping: AtomicBool,
+    /// When an event pass last ran every housekeeping step, in ns.
+    housekeeping_ns: AtomicU64,
     #[pin]
     poll: DelayedWork<Shared, POLL_WORK>,
     #[pin]
@@ -989,8 +995,17 @@ impl Shared {
         self.port.store(core::ptr::null_mut(), Ordering::Release);
     }
 
-    /// Queues the event worker. Callable from any context.
+    /// Queues the event worker for a pass that runs every housekeeping step. Callable from any
+    /// context.
     fn queue_events(self: &Arc<Self>) {
+        self.housekeeping.store(true, Ordering::Release);
+        self.queue_event_records();
+    }
+
+    /// Queues the event worker for firmware records alone: completions, or records that arrived
+    /// during a pass. The pass skips the housekeeping that nothing needs (see
+    /// `runtime_event_turn`). Callable from any context.
+    fn queue_event_records(self: &Arc<Self>) {
         // Already queued if this fails, which is just as good.
         let _ = workqueue::system_highpri().enqueue::<Arc<Shared>, EVENT_WORK>(self.clone());
     }
@@ -1170,6 +1185,8 @@ impl Gpu {
                 renders,
                 compute_demand: AtomicU32::new(0),
                 work <- new_work!("g17::Shared::work"),
+                housekeeping: AtomicBool::new(true),
+                housekeeping_ns: AtomicU64::new(0),
                 poll <- new_delayed_work!("g17::Shared::poll"),
                 idle <- new_delayed_work!("g17::Shared::idle"),
                 grow <- new_work!("g17::Shared::grow"),
diff --git a/drivers/gpu/drm/asahi/g17/object.rs b/drivers/gpu/drm/asahi/g17/object.rs
index f1b22ad5d73310..87fdb0e8a62ad1 100644
--- a/drivers/gpu/drm/asahi/g17/object.rs
+++ b/drivers/gpu/drm/asahi/g17/object.rs
@@ -466,8 +466,17 @@ struct PoolInner {
     reserved: [usize; POOL_FAMILIES],
     bytes: usize,
     stopped: bool,
+    /// Objects the firmware may still reference after their owner dropped them, kept mapped
+    /// and out of reuse until firmware stop (see [`PooledObject::retain_until_stop`]).
+    kept: KVec<KernelObject>,
+    kept_bytes: usize,
+    /// Kept objects still owned by their users, whose list entries are reserved.
+    kept_reserved: usize,
 }
 
+/// Retained objects past this many are still retained, with one warning.
+const KEPT_WARN: usize = 256;
+
 /// Per-device cache-safe object recycling. Every converted mapping reserves
 /// one return slot before its page-table attributes change. Active leases and
 /// poisoned mappings count against the same capacity as reusable objects.
@@ -490,6 +499,9 @@ struct CacheLease {
 pub(crate) struct PooledObject {
     object: core::mem::ManuallyDrop<KernelObject>,
     lease: Option<CacheLease>,
+    /// The pool keeping this object until firmware stop, and whether `reserve_kept` took its
+    /// list entry.
+    keep: Option<(kernel::sync::Arc<Pool>, bool)>,
 }
 
 impl core::ops::Deref for PooledObject {
@@ -503,11 +515,28 @@ impl core::ops::DerefMut for PooledObject {
         &mut self.object
     }
 }
+impl PooledObject {
+    /// The firmware may still reference this object after its owner drops it: keep it mapped
+    /// and never hand it out again before firmware stop. It leaves the reuse cache, so it
+    /// does not take cache capacity. Call from sleepable context: this reserves the list
+    /// entry, so the drop, wherever it happens, does not allocate.
+    pub(crate) fn retain_until_stop(&mut self, pool: &kernel::sync::Arc<Pool>) {
+        if self.keep.is_none() {
+            let reserved = pool.reserve_kept();
+            self.keep = Some((pool.clone(), reserved));
+        }
+    }
+}
+
 impl Drop for PooledObject {
     fn drop(&mut self) {
         // SAFETY: This destructor is the only operation that moves `object`
         // out; ManuallyDrop prevents a second implicit drop of the field.
         let object = unsafe { core::mem::ManuallyDrop::take(&mut self.object) };
+        if let Some((pool, reserved)) = self.keep.take() {
+            pool.keep_object(object, self.lease.take(), reserved);
+            return;
+        }
         match self.lease.take() {
             Some(lease) => lease.pool.return_object(object, &lease),
             None => drop(object),
@@ -523,7 +552,10 @@ impl Pool {
         }
         kernel::sync::Arc::pin_init(
             pin_init!(Self {
-                inner <- kernel::new_mutex!(PoolInner { returned, reserved: [0; POOL_FAMILIES], bytes: 0, stopped: false }, "g17::object::Pool"),
+                inner <- kernel::new_mutex!(PoolInner {
+                    returned, reserved: [0; POOL_FAMILIES], bytes: 0, stopped: false, kept: KVec::new(),
+                    kept_bytes: 0, kept_reserved: 0,
+                }, "g17::object::Pool"),
             }),
             GFP_KERNEL,
         )
@@ -572,6 +604,7 @@ impl Pool {
             return Ok(PooledObject {
                 object: core::mem::ManuallyDrop::new(entry.object),
                 lease: Some(lease()),
+                keep: None,
             });
         }
         let object = alloc.kernel(size, mmu::UAT_PGSZ as u64, prot, cpu)?;
@@ -596,6 +629,7 @@ impl Pool {
         let mut result = PooledObject {
             object: core::mem::ManuallyDrop::new(object),
             lease: reserved.then(lease),
+            keep: None,
         };
         if !reserved {
             return Ok(result);
@@ -655,10 +689,67 @@ impl Pool {
         core::mem::forget(object);
     }
 
+    /// Room for one more kept object, taken by the matching `keep_object`. Returns whether
+    /// it was reserved; after a failed allocation that drop leaks the object instead.
+    fn reserve_kept(&self) -> bool {
+        let mut inner = self.inner.lock();
+        let pending = inner.kept_reserved + 1;
+        if inner.kept.capacity() - inner.kept.len() < pending
+            && inner.kept.reserve(pending, GFP_KERNEL).is_err()
+        {
+            return false;
+        }
+        inner.kept_reserved = pending;
+        true
+    }
+
+    fn keep_object(&self, object: KernelObject, lease: Option<CacheLease>, reserved: bool) {
+        let mut inner = self.inner.lock();
+        if reserved {
+            inner.kept_reserved = inner.kept_reserved.saturating_sub(1);
+        }
+        if let Some(lease) = &lease {
+            // The object leaves the reuse cache and its capacity.
+            inner.reserved[lease.family as usize] -= 1;
+            inner.bytes -= object.size();
+        }
+        if inner.stopped {
+            drop(inner);
+            drop(object);
+            return;
+        }
+        let size = object.size();
+        if inner.kept.len() == KEPT_WARN {
+            pr_warn!("G17 keeps {} firmware-linked objects until the GPU firmware restarts\n", KEPT_WARN);
+        }
+        if !reserved {
+            // Another kept object's reservation must not be used: leak this one instead.
+            pr_err!("G17 could not record a firmware-linked object; leaking it\n");
+            core::mem::forget(object);
+            return;
+        }
+        // retain_until_stop reserved the entry; this never allocates.
+        match inner.kept.push_within_capacity(object) {
+            Ok(()) => inner.kept_bytes += size,
+            Err(error) => {
+                // Never unmap an object the firmware may still reference.
+                pr_err!("G17 could not record a firmware-linked object; leaking it\n");
+                core::mem::forget(error.0);
+            }
+        }
+    }
+
     /// Releases retained mappings after both firmware processors have stopped.
     /// Active objects returned later observe the same stop witness.
     pub(crate) fn stop_after_firmware(&self) {
-        self.inner.lock().stopped = true;
+        let kept = {
+            let mut inner = self.inner.lock();
+            inner.stopped = true;
+            inner.kept_bytes = 0;
+            core::mem::take(&mut inner.kept)
+        };
+        // Mapping teardown must not run while the pool mutex is held.
+        drop(kept);
         for index in 0..POOL_FAMILIES * POOL_SLOTS {
             let entry = {
                 let mut inner = self.inner.lock();
@@ -684,6 +775,7 @@ impl Pool {
                      inner.reserved[family], POOL_SLOTS, entries.iter().flatten().count(),
                      entries.iter().flatten().filter(|entry| entry.poisoned).count())?;
         }
-        writeln!(f, "cached_object_bytes used={} total={}", inner.bytes, POOL_BYTES)
+        writeln!(f, "cached_object_bytes used={} total={}", inner.bytes, POOL_BYTES)?;
+        writeln!(f, "firmware_linked_objects kept={} bytes={}", inner.kept.len(), inner.kept_bytes)
     }
 }
diff --git a/drivers/gpu/drm/asahi/g17/preparation.rs b/drivers/gpu/drm/asahi/g17/preparation.rs
index eb2366fa3451b2..2aa32131150767 100644
--- a/drivers/gpu/drm/asahi/g17/preparation.rs
+++ b/drivers/gpu/drm/asahi/g17/preparation.rs
@@ -60,7 +60,7 @@ pub(super) struct Gate {
 }
 
 impl Gate {
-    #[cfg(CONFIG_KUNIT)]
+    #[cfg(CONFIG_KUNIT = "y")]
     pub(super) fn new() -> Result<Arc<Self>> {
         Self::with_waiters(None)
     }
diff --git a/drivers/gpu/drm/asahi/g17/queue/admission.rs b/drivers/gpu/drm/asahi/g17/queue/admission.rs
index e1e7552c243f24..ed73323e34e6ff 100644
--- a/drivers/gpu/drm/asahi/g17/queue/admission.rs
+++ b/drivers/gpu/drm/asahi/g17/queue/admission.rs
@@ -71,11 +71,19 @@ struct State {
     /// their own fence with EBUSY instead of a device-loss error, so the VM stays usable.
     spared: u64,
     kill_hold: u64,
+    /// Slots whose kill hold ended at their killed context's consumed scheduler release,
+    /// and when. A new owner takes such a pair last, the one released longest ago first.
+    killed: u64,
+    killed_at: [u64; cfg::RENDER_SLOTS],
+    /// The physical queue table still has room for a pair built in an unowned slot. The
+    /// runtime updates this whenever the table changes.
+    build_room: bool,
     released: u64,
     epoch: u64,
     footprints: KVec<InFlight>,
     next_ticket: u64,
 }
+
 impl State {
     const MASK: u64 = (u64::MAX >> (64 - cfg::RENDER_SLOTS)) & !1;
     const fn new() -> Self {
@@ -87,6 +95,9 @@ impl State {
             quarantined: 0,
             spared: 0,
             kill_hold: 0,
+            killed: 0,
+            killed_at: [0; cfg::RENDER_SLOTS],
+            build_room: true,
             released: 0,
             epoch: 0,
             footprints: KVec::new(),
@@ -120,6 +131,15 @@ impl State {
         self.epochs[slot] = self.epoch;
     }
     fn take_idle(&mut self, slot: usize, owner: u64) -> u8 {
+        if self.owners[slot] != owner {
+            crate::g17::trace::teardown(
+                6,
+                slot as u64,
+                self.owners[slot],
+                owner | u64::from(self.killed & (1u64 << slot) != 0) << 63,
+            );
+            self.killed &= !(1u64 << slot);
+        }
         self.refs[slot] += 1;
         self.active |= 1u64 << slot;
         self.released &= !(1u64 << slot);
@@ -165,13 +185,23 @@ impl State {
         }
         let free = !self.active & Self::MASK & !self.quarantined & !self.kill_hold;
         let eligible = |slot: usize| free & (1u64 << slot) != 0;
-        // Reuse a closed, exactly retired owner before consuming a new graph.
+        let released = |slot: usize| self.released & (1u64 << slot) != 0;
+        let unowned = || (0..cfg::RENDER_SLOTS).find(|slot| eligible(*slot) && self.owners[*slot] == 0);
+        // Reuse a closed, exactly retired owner before consuming a new graph. A killed
+        // context's pair comes after a new graph while one can still be built: on Neo, new
+        // owners of such pairs had renders that executed but never stored their stamps.
+        // Once the queue table is full it is reused, the one released longest ago first,
+        // so killed pairs never strand physical queues.
         let slot = (0..cfg::RENDER_SLOTS)
-            .filter(|slot| eligible(*slot) && self.released & (1u64 << slot) != 0)
+            .filter(|slot| eligible(*slot) && released(*slot) && self.killed & (1u64 << slot) == 0)
             .min_by_key(|slot| self.epochs[*slot])
+            .or_else(|| if self.build_room { unowned() } else { None })
             .or_else(|| {
-                (0..cfg::RENDER_SLOTS).find(|slot| eligible(*slot) && self.owners[*slot] == 0)
-            })?;
+                (0..cfg::RENDER_SLOTS)
+                    .filter(|slot| eligible(*slot) && released(*slot))
+                    .min_by_key(|slot| self.killed_at[*slot])
+            })
+            .or_else(unowned)?;
         let first = self.take_idle(slot, owner);
         self.claim_sibling(owner);
         Some(first)
@@ -180,7 +210,8 @@ impl State {
         if self.owners.iter().filter(|id| **id == owner).count() >= cfg::RENDER_PAIRS_PER_OWNER {
             return;
         }
-        let free = !self.active & Self::MASK & !self.quarantined & !self.kill_hold;
+        // A speculative sibling never takes a killed context's pair.
+        let free = !self.active & Self::MASK & !self.quarantined & !self.kill_hold & !self.killed;
         let eligible = |slot: usize| free & (1u64 << slot) != 0;
         let sibling = (0..cfg::RENDER_SLOTS)
             .filter(|slot| {
@@ -190,6 +221,10 @@ impl State {
             })
             .min_by_key(|slot| self.epochs[*slot])
             .or_else(|| {
+                // Nor an unowned slot once no pair can be built there.
+                if !self.build_room {
+                    return None;
+                }
                 (0..cfg::RENDER_SLOTS).find(|slot| eligible(*slot) && self.owners[*slot] == 0)
             });
         if let Some(slot) = sibling {
@@ -223,6 +258,7 @@ impl State {
         let bit = 1u64 << slot;
         let old = self.owners[slot];
         if old == 0
+            || self.killed & bit != 0
             || owner == Some(old)
             || expected.is_some_and(|expected| expected != old)
             || self.refs[slot] != 0
@@ -406,7 +442,7 @@ pub(crate) struct Pool {
     renders: Option<Arc<crate::g17::availability::Availability>>,
 }
 impl Pool {
-    #[cfg(CONFIG_KUNIT)]
+    #[cfg(CONFIG_KUNIT = "y")]
     pub(crate) fn new(max_queues: u32) -> Result<Arc<Self>> {
         Self::with_waiters(max_queues, None, None)
     }
@@ -648,9 +684,27 @@ impl Pool {
         state.quarantined &= !released;
         state.spared &= !released;
         state.kill_hold = 0;
+        state.killed = 0;
         drop(state);
         self.returned();
     }
+
+    /// Whether the physical queue table can hold another pair (see `State::build_room`).
+    pub(crate) fn note_build_room(&self, room: bool) {
+        let mut state = self.state.lock();
+        if state.build_room != room {
+            state.build_room = room;
+            drop(state);
+            self.returned();
+        }
+    }
+
+    /// A killed context's pair is free for a new owner.
+    pub(crate) fn killed_pair_free(&self) -> bool {
+        let state = self.state.lock();
+        state.killed & state.released & !state.active & !state.quarantined & !state.kill_hold & State::MASK != 0
+    }
+
 }
 
 #[kunit_tests(asahi_g17_admission_accounting)]
@@ -863,7 +917,7 @@ impl Drop for RenderLease {
 }
 
 impl State {
-    fn hold_for_exit_kill(&mut self, owner: u64, slots: u64) -> bool {
+    fn hold_for_exit_kill(&mut self, owner: u64, slots: u64, executed: u64) -> bool {
         if owner == 0 || slots & !Self::MASK != 0 {
             return false;
         }
@@ -872,10 +926,10 @@ impl State {
             if slots & bit == 0 {
                 continue;
             }
+            // A pair whose commands all ran is held with them (see `Pair::executed_for_kill`).
             if self.owners[slot] != owner
-                || self.refs[slot] != 0
-                || self.active & bit != 0
-                || self.released & bit == 0
+                || (executed & bit == 0
+                    && (self.refs[slot] != 0 || self.active & bit != 0 || self.released & bit == 0))
             {
                 return false;
             }
@@ -886,12 +940,13 @@ impl State {
 }
 impl Pool {
     /// Hold the departed owner's idle pairs before publishing its exit record.
-    /// Successful publication retains these bits until both processors stop.
-    pub(crate) fn hold_render_slots_for_exit_kill(&self, owner: u64, slots: u64) -> bool {
-        self.state.lock().hold_for_exit_kill(owner, slots)
+    /// Successful publication retains these bits until the owner's scheduler release is
+    /// consumed (`release_killed_slots` at detach), or until both processors stop.
+    pub(crate) fn hold_render_slots_for_exit_kill(&self, owner: u64, slots: u64, executed: u64) -> bool {
+        self.state.lock().hold_for_exit_kill(owner, slots, executed)
     }
 
-    /// Undo only a hold whose control record was not published.
+    /// Undo a hold whose control record was not published.
     pub(crate) fn release_exit_kill_hold(&self, slots: u64) {
         if slots == 0 {
             return;
@@ -899,6 +954,26 @@ impl Pool {
         self.state.lock().kill_hold &= !slots;
         self.returned();
     }
+
+    /// End the kill hold of `slots` once their killed context's scheduler release was
+    /// consumed (the firmware no longer names these pairs). A new owner takes them last.
+    pub(crate) fn release_killed_slots(&self, slots: u64) {
+        let mut state = self.state.lock();
+        let held = state.kill_hold & slots & State::MASK;
+        if held == 0 {
+            return;
+        }
+        let now = crate::g17::now_ns();
+        state.kill_hold &= !held;
+        state.killed |= held;
+        for slot in 0..cfg::RENDER_SLOTS {
+            if held & (1u64 << slot) != 0 {
+                state.killed_at[slot] = now;
+            }
+        }
+        drop(state);
+        self.returned();
+    }
 }
 
 impl Pool {
@@ -972,13 +1047,86 @@ mod fault_capacity_tests {
         let bit = 1u64 << lease.slot();
         pool.release_owner(1);
         drop(lease);
-        assert!(pool.hold_render_slots_for_exit_kill(1, bit));
+        assert!(pool.hold_render_slots_for_exit_kill(1, bit, 0));
         assert!(!pool.reassign_unused_spare(bit.trailing_zeros() as u8, Some(1), Some(2)));
         // A failed (unpublished) control may roll back its hold. Published kills
         // keep the existing processor-stop rule; this test must not weaken it.
         pool.release_retained_owners_after_processor_stop();
         assert_eq!(pool.state.lock().kill_hold, 0);
     }
+
+    #[test]
+    fn consumed_scheduler_release_ends_the_kill_hold() {
+        let pool = Pool::new(MAX_EXECUTION_CONTEXTS).unwrap();
+        let lease = pool.acquire_render(1, None, u64::MAX, || Ok(())).unwrap();
+        let slot = lease.slot();
+        let bit = 1u64 << slot;
+        pool.release_owner(1);
+        drop(lease);
+        assert!(pool.hold_render_slots_for_exit_kill(1, bit, 0));
+        assert!(!pool.reassign_unused_spare(slot, Some(1), Some(2)));
+        pool.release_killed_slots(bit);
+        assert_eq!(pool.state.lock().kill_hold, 0);
+        assert_eq!(pool.state.lock().killed, bit);
+        assert!(pool.killed_pair_free());
+        // The killed context's pair is never reassigned as a spare, and a new owner takes
+        // a new graph before it.
+        assert!(!pool.reassign_unused_spare(slot, Some(1), Some(2)));
+        let next = pool.acquire_render(2, None, u64::MAX, || Ok(())).unwrap();
+        assert_ne!(next.slot(), slot);
+        drop(next);
+        pool.release_retained_owners_after_processor_stop();
+        assert_eq!(pool.state.lock().killed, 0);
+    }
+
+    #[test]
+    fn killed_pairs_are_taken_last_oldest_first() {
+        let pool = Pool::new(MAX_EXECUTION_CONTEXTS).unwrap();
+        {
+            let mut state = pool.state.lock();
+            for slot in 1..cfg::RENDER_SLOTS {
+                state.owners[slot] = 100 + slot as u64;
+                state.released |= 1u64 << slot;
+                state.killed |= 1u64 << slot;
+                state.killed_at[slot] = 1000 - slot as u64;
+            }
+            // One pair released without a kill.
+            state.killed &= !(1u64 << 7);
+        }
+        let first = pool.acquire_render(1, None, u64::MAX, || Ok(())).unwrap();
+        assert_eq!(first.slot(), 7);
+        let second = pool.acquire_render(2, None, u64::MAX, || Ok(())).unwrap();
+        assert_eq!(usize::from(second.slot()), cfg::RENDER_SLOTS - 1);
+        assert_eq!(pool.state.lock().killed & (1u64 << second.slot()), 0);
+    }
+
+    #[test]
+    fn a_full_queue_table_reuses_killed_pairs_before_unowned_slots() {
+        let pool = Pool::new(MAX_EXECUTION_CONTEXTS).unwrap();
+        {
+            let mut state = pool.state.lock();
+            for slot in 1..8 {
+                state.owners[slot] = 100 + slot as u64;
+                state.released |= 1u64 << slot;
+                state.killed |= 1u64 << slot;
+                state.killed_at[slot] = 1000 - slot as u64;
+            }
+        }
+        // With room in the queue table a new owner builds a pair in an unowned slot.
+        let built = pool.acquire_render(1, None, u64::MAX, || Ok(())).unwrap();
+        assert!(usize::from(built.slot()) >= 8);
+        // Without room it takes the killed pair released longest ago.
+        pool.note_build_room(false);
+        let reused = pool.acquire_render(2, None, u64::MAX, || Ok(())).unwrap();
+        assert_eq!(reused.slot(), 7);
+        assert_eq!(pool.state.lock().killed & (1u64 << 7), 0);
+        // It claims no sibling: neither another killed pair nor an unowned slot.
+        let state = pool.state.lock();
+        for slot in 1..7 {
+            assert_eq!(state.owners[slot], 100 + slot as u64);
+        }
+        assert_eq!(state.owners.iter().filter(|owner| **owner == 2).count(), 1);
+    }
 }
 
 impl Pool {
diff --git a/drivers/gpu/drm/asahi/g17/queue/compute.rs b/drivers/gpu/drm/asahi/g17/queue/compute.rs
index 9c4c5c92d049df..e1bda516ee1022 100644
--- a/drivers/gpu/drm/asahi/g17/queue/compute.rs
+++ b/drivers/gpu/drm/asahi/g17/queue/compute.rs
@@ -566,6 +566,9 @@ pub(crate) struct Queue {
     replay_error: Option<Error>,
     replay_prepared: bool,
     exit: Option<exit::Transaction>,
+    /// Since when the kill of this failed queue's context waits for other contexts' compute,
+    /// and whether that wait was reported.
+    kill_wait: Option<(u64, bool)>,
     installed: bool,
     /// Bit `p` set once a work slot of priority class `p` named this queue.
     outer_started: u8,
@@ -575,6 +578,9 @@ pub(crate) struct Queue {
     config_cache: Option<(QueueConfig, u64)>,
     ordinal: u32,
     submitted: u32,
+    /// The stamp of the newest command the firmware reported failed ([`Self::note_fault`]),
+    /// while it is newer than the stored stamp.
+    fault_stamp: Option<u32>,
     /// Prepared packets of the bound owner about to publish (see [`Self::claim`]).
     claims: u32,
     last_published: u64,
@@ -594,8 +600,9 @@ pub(crate) struct Queue {
     source_readers: Arc<core::sync::atomic::AtomicU32>,
     source_hint: crate::g17::sync_input::ReaderHint,
     /// The doorbell withheld from publications that had [`WITHHOLD_DEPTH`] - 1 or more
-    /// unretired commands of this queue ahead of them, the publication of the first, and the
-    /// compute work slots written at its priority once its slot was.
+    /// unretired commands of this queue ahead of them, the publication of the first of them
+    /// that no doorbell has announced, and the compute work slots written at its priority once
+    /// the last one's slot was.
     withheld_doorbell: Option<(u64, u64, u64)>,
 }
 
@@ -628,7 +635,10 @@ impl Queue {
     }
 
     /// The queue has left ordinary retirement: a timeout, kill, quarantine, replay, recovery
-    /// witness or exit is in progress.
+    /// witness or exit is in progress, or retirement stopped at a command the firmware
+    /// finished without a completion record. A front whose record has arrived and that only
+    /// waits for its stamp is ordinary retirement: [`Self::due_doorbell`] counts it as
+    /// finished, and ringing at once there rang most withheld doorbells early.
     fn settling(&self) -> bool {
         self.quarantined
             || self.released
@@ -636,7 +646,8 @@ impl Queue {
             || self.replays.len != 0
             || self.replay_retirement
             || self.awaiting_witness
-            || self.retire_pending
+            || (self.retire_pending
+                && self.active.front().is_none_or(|active| active.observed.is_none()))
             || self.retired_by_teardown
             || self.spared_quarantine
             || self.failure_status_pending
@@ -688,12 +699,14 @@ impl Queue {
             replay_error: None,
             replay_prepared: false,
             exit: None,
+            kill_wait: None,
             installed: false,
             outer_started: 0,
             published_config: None,
             config_cache: None,
             ordinal: 0,
             submitted: 0,
+            fault_stamp: None,
             claims: 0,
             last_published: 0,
             last_end: 0,
@@ -714,6 +727,37 @@ impl Queue {
             withheld_doorbell: None,
         })
     }
+    /// The firmware failed the command whose completion stamp is `stamp`. It writes a
+    /// zero-duration completion record for it, which quarantines the queue, but stores no
+    /// stamp: without this the quarantine would wait for its retirement witness forever and
+    /// keep the queue's backing. Commands published after the failed one still need their own
+    /// stamp or record.
+    pub(crate) fn note_fault(&mut self, stamp: u32) {
+        if self
+            .fault_stamp
+            .is_none_or(|old| retirement::stamp_newer(stamp, old))
+        {
+            self.fault_stamp = Some(stamp);
+        }
+    }
+
+    /// The firmware failed a command of this quarantined queue, and the queue still waits for
+    /// its retirement witness: commands published after the failed one, which the firmware
+    /// never runs, or the failed command itself, whose ring entry it does not consume (a ray
+    /// traversal's, on Neo). Only a kill of the queue's context gives them back.
+    pub(crate) fn fault_pending(&self) -> bool {
+        self.fault_stamp.is_some() && self.quarantined && self.awaiting_witness && self.exit.is_none()
+    }
+
+    /// The stamp that proves retirement: the stored one, or a newer reported fault's.
+    fn witness_stamp(&self) -> Result<u32> {
+        let stored = self.graph.stamp()?;
+        Ok(match self.fault_stamp {
+            Some(fault) if retirement::stamp_newer(fault, stored) => fault,
+            _ => stored,
+        })
+    }
+
     pub(crate) fn qid(&self) -> u8 {
         self.kick.id().qid()
     }
@@ -747,6 +791,16 @@ impl Queue {
     pub(crate) fn in_flight(&self) -> bool {
         self.active.len != 0 && !self.quarantined
     }
+    /// The oldest command is a published long job (ray traversal or AS build) whose
+    /// completion record the driver has not seen yet.
+    pub(crate) fn long_job_running(&self) -> bool {
+        !self.quarantined
+            && self.active.front().is_some_and(|active| {
+                active.observed.is_none()
+                    && active.packet.is_published()
+                    && active.packet.command.long_job()
+            })
+    }
     /// Quarantines without a pending witness no longer describe live pipe order.
     pub(crate) fn timeout_head(&self) -> Option<crate::g17::timeout::Head> {
         if self.quarantined && !self.awaiting_witness {
@@ -1096,7 +1150,7 @@ impl Queue {
             if !retirement::quarantined_retired(
                 &newest,
                 self.graph.cursors()?,
-                self.graph.stamp()?,
+                self.witness_stamp()?,
                 self.active.len as u32,
                 self.submitted,
             )? {
@@ -1140,6 +1194,7 @@ impl Queue {
             drop(self.active.pop());
         }
         self.awaiting_witness = false;
+        self.fault_stamp = None;
         self.retirement_proved = false;
         self.replay_retirement = false;
         self.spared_deferred = false;
@@ -1584,6 +1639,7 @@ impl Queue {
         self.spared_deferred = false;
         self.retire_pending = false;
         self.awaiting_witness = false;
+        self.fault_stamp = None;
         self.retirement_proved = false;
         self.replay_retirement = false;
         self.replay_error = None;
@@ -1952,6 +2008,8 @@ pub(crate) trait Host {
         -> Result;
     fn notify_compute(&mut self, priority: u8, message: u64) -> Result;
     fn compute_written(&self, priority: u8) -> u64;
+    /// Compute work slots at `priority` that a doorbell has announced.
+    fn compute_announced(&self, priority: u8) -> u64;
     fn note_submission(&mut self) -> Result;
 }
 
@@ -2303,6 +2361,15 @@ impl Queue {
                 && self.withheld_doorbell.is_none_or(|(withheld, _, _)| withheld == doorbell)
             {
                 let written = host.compute_written(policy.priority());
+                // Once another doorbell announced every slot the withheld one covered, it
+                // starts over from this publication: it then waits for this command's own
+                // predecessors rather than for those of slots already announced.
+                if self
+                    .withheld_doorbell
+                    .is_some_and(|(_, _, last)| host.compute_announced(policy.priority()) >= last)
+                {
+                    self.withheld_doorbell = None;
+                }
                 let entry = self.withheld_doorbell.get_or_insert((doorbell, published, written));
                 entry.2 = written;
             } else {
diff --git a/drivers/gpu/drm/asahi/g17/queue/compute/exit.rs b/drivers/gpu/drm/asahi/g17/queue/compute/exit.rs
index d7386e335e1671..719a74787ca706 100644
--- a/drivers/gpu/drm/asahi/g17/queue/compute/exit.rs
+++ b/drivers/gpu/drm/asahi/g17/queue/compute/exit.rs
@@ -83,11 +83,12 @@ impl Transaction {
 }
 
 impl Queue {
-    /// A command with ray traversal scratch is never killed. On Neo, killing a
+    /// A running command with ray traversal scratch is never killed. On Neo, killing a
     /// traversal that stopped making progress either crashed the GPU firmware
     /// when the queue's pool was then released, or stalled every later render
     /// when it was retained; left alone, the rest of the GPU kept working. Such
-    /// a queue stays quarantined and retained until its commands retire.
+    /// a queue stays quarantined and retained until its commands retire. A
+    /// traversal the firmware reported failed no longer runs (see `fault_pending`).
     fn active_traversal(&self) -> bool {
         self.active.iter().any(|active| {
             matches!(
@@ -101,7 +102,7 @@ impl Queue {
             && self.released
             && self.active.len != 0
             && self.exit.is_none()
-            && !self.active_traversal()
+            && (!self.active_traversal() || self.fault_pending())
             && (self.quarantined || self.retired_by_teardown)
             && age
                 >= if self.quarantined {
@@ -110,6 +111,10 @@ impl Queue {
                     CANCELLED_DELAY_NS
                 }
     }
+    /// A kill of `context` was published through this queue.
+    pub(crate) fn exit_published(&self, context: &Arc<Context>) -> bool {
+        self.exit.is_some() && self.context().is_some_and(|bound| Arc::ptr_eq(bound, context))
+    }
     pub(crate) fn exit_owner(&self, context: &Arc<Context>) -> Option<u64> {
         if !self
             .active
@@ -124,6 +129,23 @@ impl Queue {
                 .and_then(|active| active.ticket.map(|ticket| ticket.owner))
         })
     }
+    /// The kill of this queue's failed context waits; returns how long it has waited, and
+    /// whether this is the first time it crossed `slow`.
+    pub(crate) fn note_kill_wait(&mut self, now: u64, slow: u64) -> (u64, bool) {
+        let (since, warned) = self.kill_wait.get_or_insert((now, false));
+        let waited = now.saturating_sub(*since);
+        let crossed = waited >= slow && !*warned;
+        *warned |= crossed;
+        (waited, crossed)
+    }
+    /// How long the kill of this queue's failed context has been waiting, if it is.
+    pub(crate) fn kill_waiting(&self, now: u64) -> Option<u64> {
+        self.kill_wait.map(|(since, _)| now.saturating_sub(since))
+    }
+    /// The kill was published: how long it waited, if it did.
+    pub(crate) fn take_kill_wait(&mut self, now: u64) -> u64 {
+        self.kill_wait.take().map_or(0, |(since, _)| now.saturating_sub(since))
+    }
     pub(crate) fn record_exit(&mut self, cookie: u64, now: u64) {
         self.exit = Some(Transaction {
             cookie,
@@ -162,6 +184,7 @@ impl Queue {
             return false;
         }
         exit.phase = Phase::Acknowledged;
+        crate::g17::trace::teardown(4, cookie, u64::from(self.qid()), 0);
         true
     }
     pub(crate) fn settle_exit(
diff --git a/drivers/gpu/drm/asahi/g17/queue/compute/retirement.rs b/drivers/gpu/drm/asahi/g17/queue/compute/retirement.rs
index c2b49e7b436d77..8a737546cb7cf9 100644
--- a/drivers/gpu/drm/asahi/g17/queue/compute/retirement.rs
+++ b/drivers/gpu/drm/asahi/g17/queue/compute/retirement.rs
@@ -52,6 +52,12 @@ impl Cursors {
     }
 }
 
+/// Whether stamp `a` was stored after stamp `b` (both kick stamps, modulo wrap).
+pub(crate) const fn stamp_newer(a: u32, b: u32) -> bool {
+    let delta = a.wrapping_sub(b);
+    delta != 0 && delta < 0x8000_0000
+}
+
 /// Outstanding kicks span fewer than half the 24-bit stamp sequence. Zero is a valid wrap.
 pub(crate) const fn stamp_covers(word: u32, kick: u64) -> bool {
     let delta = word.wrapping_sub((kick as u32).wrapping_shl(8));
@@ -447,6 +453,15 @@ mod tests {
         assert!(stamp_covers(0, 0xff_ffff));
     }
 
+    #[test]
+    fn a_reported_fault_stamp_is_newer_only_until_stored() {
+        assert!(stamp_newer(own(KICK), own(KICK - 1)));
+        assert!(!stamp_newer(own(KICK), own(KICK)));
+        assert!(!stamp_newer(own(KICK - 1), own(KICK)));
+        // Across the wrap of the 24-bit kick in the stamp word.
+        assert!(stamp_newer(own(0), own(0xff_ffff)));
+    }
+
     #[test]
     fn a_late_stamp_is_waited_for() {
         // Under games on J700 the stamp lands up to about 4 ms after the record while the
diff --git a/drivers/gpu/drm/asahi/g17/queue/render.rs b/drivers/gpu/drm/asahi/g17/queue/render.rs
index 696a1486eb0ec5..3261f2f0640b32 100644
--- a/drivers/gpu/drm/asahi/g17/queue/render.rs
+++ b/drivers/gpu/drm/asahi/g17/queue/render.rs
@@ -1034,11 +1034,7 @@ impl Pair {
         ]
     }
     fn selected(&self, masks: Option<[u64; 2]>) -> bool {
-        masks.is_none_or(|masks| {
-            self.qids()
-                .into_iter()
-                .all(|qid| masks[usize::from(qid) / 64] & (1u64 << (qid % 64)) != 0)
-        })
+        masks.is_none_or(|masks| completion_noted(self.qids(), masks))
     }
     /// The oldest published group still running, for cross-VM timeout attribution
     /// (crate::g17::timeout). A quarantined pair's work holds the GPU until the firmware drains
@@ -1614,7 +1610,43 @@ impl Pair {
         let counters = self.free_list.counters()?;
         let stamps = host.stamps(self.qids())?;
         let (_, state) = host.epoch()?;
-        Ok(retirement::drained(self.ordinal, stamps, counters, state))
+        Ok(retirement::drained(self.ordinal, stamps, counters, state)
+            || (state == 0 && self.executed_unstamped() && self.context_killed()))
+    }
+
+    /// The firmware acknowledged a kill of this pair's context (scheduler page): it no longer
+    /// walks that context's work list, so the commands it never unlinked can be released.
+    fn context_killed(&self) -> bool {
+        crate::g17::queue::compute::exit::State::read(&self.memory.context).is_ok_and(|state| state.killed != 0)
+    }
+
+    /// The pair's context may be killed although its commands were not retired: they all ran
+    /// (`executed_unstamped`), so the kill leaves nothing queued for the firmware to run.
+    /// Its memory stays until the kill is acknowledged.
+    pub(crate) fn executed_for_kill(&self, context: &Arc<Context>) -> bool {
+        Arc::ptr_eq(context, &self.memory.context) && self.executed_unstamped()
+    }
+
+    /// Every command of this quarantined pair ran, its ring entries were consumed and its
+    /// free-list blocks returned, yet the firmware never stores the stamp of the last one.
+    /// On Neo this is the render work of a context whose compute command failed: it runs
+    /// about a second after the failure, but its stamp never arrives and it stays linked in
+    /// the firmware's job list, so neither the stamp proof nor the abandoned-group proof
+    /// ever holds and the pair would stay quarantined, keeping its context, until the
+    /// processors stop.
+    fn executed_unstamped(&self) -> bool {
+        self.quarantined
+            // Only for a closed owner: a live owner would go on publishing into a context
+            // whose work list still links the unstamped node.
+            && self.owner.is_none()
+            && self.prepared.is_none()
+            && self.active.len() != 0
+            && self.free_list.counters().is_ok_and(|counters| counters[0] == counters[1])
+            && self.active.iter().all(|active| {
+                active.outer_published
+                    && observation::snapshot(&self.memory, active.ticket.ordinal)
+                        .is_ok_and(retirement::Snapshot::executed_and_consumed)
+            })
     }
 
     /// Account one witnessed tracker, preserving all completion ownership.
@@ -1628,6 +1660,10 @@ impl Pair {
         let qids = self.qids();
         let qos = self.qos_owners();
         let buffer = self.buffer_id();
+        // Commands the firmware never unlinked are released only after it acknowledged the
+        // kill of their context, on every path (see `firmware_drained`).
+        let unstamped = self.executed_unstamped();
+        let killed = unstamped && self.context_killed();
         let Some(active) = self.active.iter_mut().find(|a| !a.flist_completed) else {
             return Ok(false);
         };
@@ -1638,6 +1674,12 @@ impl Pair {
         };
         let result = match retirement::poll(&mut active.ticket, &mut observer)? {
             retirement::Poll::Complete(ticks) => Ok(ticks),
+            // Its commands ran without a stamp (see `executed_unstamped`): they already failed
+            // with the quarantine, and the client sees them cancelled.
+            retirement::Poll::Pending | retirement::Poll::Retry if killed => {
+                crate::g17::trace::teardown(2, u64::from(self.slot), active.ticket.ordinal, 0);
+                Err(ECANCELED)
+            }
             retirement::Poll::Pending | retirement::Poll::Retry => {
                 core::sync::atomic::fence(core::sync::atomic::Ordering::Acquire);
                 let snapshot = observer.snapshot(active.ticket.ordinal)?;
@@ -2200,3 +2242,46 @@ impl Pair {
         Ok(())
     }
 }
+
+/// Whether `masks`, the stamp notifications collected for a pair whose tiling and fragment
+/// queues are `qids`, can mean a finished pass: the fragment queue's bit. A fragment kick starts
+/// only after its fragment barrier saw the tiling stamp, and retirement still checks both stamps
+/// in memory, so the tiling notification is not needed to retire.
+pub(crate) fn completion_noted(qids: [u8; 2], masks: [u64; 2]) -> bool {
+    let fragment = usize::from(qids[1]);
+    masks[fragment / 64] & (1u64 << (fragment % 64)) != 0
+}
+
+#[kunit_tests(asahi_g17_render_completion_noted)]
+mod completion_noted_tests {
+    use super::completion_noted;
+
+    fn bit(qid: u8) -> [u64; 2] {
+        let mut masks = [0; 2];
+        masks[usize::from(qid) / 64] |= 1u64 << (qid % 64);
+        masks
+    }
+
+    #[test]
+    fn fragment_notification_alone_selects() {
+        assert!(completion_noted([3, 4], bit(4)));
+        assert!(completion_noted([70, 100], bit(100)));
+    }
+
+    #[test]
+    fn tiling_notification_alone_does_not_select() {
+        assert!(!completion_noted([3, 4], bit(3)));
+        assert!(!completion_noted([70, 100], bit(70)));
+        assert!(!completion_noted([3, 4], [0; 2]));
+    }
+
+    #[test]
+    fn other_queues_do_not_select() {
+        assert!(!completion_noted([3, 4], bit(5)));
+        assert!(!completion_noted([3, 68], bit(4)));
+        let mut both = bit(3);
+        both[1] |= bit(68)[1];
+        assert!(completion_noted([3, 68], both));
+    }
+}
+
diff --git a/drivers/gpu/drm/asahi/g17/queue/render/memory.rs b/drivers/gpu/drm/asahi/g17/queue/render/memory.rs
index 756f96e3fd048d..884537b99b6d83 100644
--- a/drivers/gpu/drm/asahi/g17/queue/render/memory.rs
+++ b/drivers/gpu/drm/asahi/g17/queue/render/memory.rs
@@ -528,6 +528,8 @@ impl Memory {
 }
 
 const _: () = {
+    // Every unretired render of a pair holds its own command record.
+    assert!(DEPTH <= RECORDS);
     assert!(abi::TA_DESCRIPTOR_SIZE * RECORDS <= TA_BYTES);
     assert!(abi::FRAGMENT_SLOT_SIZE * RECORDS <= FRAGMENT_BYTES);
     assert!(RECORDS * CONFIG_STRIDE <= 0x4000);
diff --git a/drivers/gpu/drm/asahi/g17/queue/render/retirement.rs b/drivers/gpu/drm/asahi/g17/queue/render/retirement.rs
index 00e4997ef3661e..7996970ee369ff 100644
--- a/drivers/gpu/drm/asahi/g17/queue/render/retirement.rs
+++ b/drivers/gpu/drm/asahi/g17/queue/render/retirement.rs
@@ -127,6 +127,12 @@ impl Snapshot {
     pub(crate) fn untouched(self) -> bool {
         self.pass == [0, 0]
     }
+
+    /// The command ran (tiler start and fragment end written) and the firmware consumed both
+    /// queues' ring entries.
+    pub(crate) fn executed_and_consumed(self) -> bool {
+        self.executed() && self.cursors.iter().all(|cursors| cursors.done == cursors.write)
+    }
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -355,3 +361,25 @@ impl<T> Tracker<T> {
             .filter_map(Option::as_mut)
     }
 }
+
+#[kunit_tests(asahi_g17_render_retirement)]
+mod executed_tests {
+    use super::*;
+
+    fn snapshot(done: u32, write: u32, pass: [u64; 2]) -> Snapshot {
+        Snapshot {
+            cursors: [Cursors { done, read: done, write }; 2],
+            job_list_empty: false,
+            pass,
+        }
+    }
+
+    #[test]
+    fn executed_and_consumed_needs_both_passes_and_consumed_rings() {
+        assert!(snapshot(3, 3, [1, 2]).executed_and_consumed());
+        assert!(!snapshot(3, 3, [0, 0]).executed_and_consumed());
+        assert!(!snapshot(3, 3, [1, 0]).executed_and_consumed());
+        assert!(!snapshot(3, 3, [0, 2]).executed_and_consumed());
+        assert!(!snapshot(2, 3, [1, 2]).executed_and_consumed());
+    }
+}
diff --git a/drivers/gpu/drm/asahi/g17/runtime.rs b/drivers/gpu/drm/asahi/g17/runtime.rs
index e04295d5766063..108c9fe19a462f 100644
--- a/drivers/gpu/drm/asahi/g17/runtime.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime.rs
@@ -186,7 +186,7 @@ pub(super) struct Registry {
 impl Registry {
     /// Initialize inside the firmware allocation instead of staging the whole
     /// registry on the probe stack. No additional allocation is required.
-    #[cfg(CONFIG_KUNIT)]
+    #[cfg(CONFIG_KUNIT = "y")]
     pub(super) fn new(
         clusters: u32,
         deflake: crate::g17::fw::render::Deflake,
@@ -541,6 +541,7 @@ impl Registry {
             .qids
             .reserve_for(identity, DataMaster::Compute, Some(FIRST_COMPUTE_QID), queues)
             .map_err(|error| self.compute_admission_error(error))?;
+        self.note_build_room();
         let reservation = ComputeReservation { slot, id, owner };
         self.compute[slot] = Some(ComputeEntry {
             reservation,
@@ -584,6 +585,7 @@ impl Registry {
         self.reserved(reservation)?;
         self.qids.cancel_unpublished(reservation.id)?;
         self.compute[reservation.slot] = None;
+        self.note_build_room();
         Ok(())
     }
 
@@ -601,6 +603,7 @@ impl Registry {
         if unpublished {
             self.qids.cancel_unpublished(entry.reservation.id)?;
             self.compute[slot] = None;
+            self.note_build_room();
         }
         Ok(())
     }
@@ -1396,6 +1399,9 @@ impl compute::Host for ComputeHost<'_> {
     fn compute_written(&self, priority: u8) -> u64 {
         self.port.compute_written(priority)
     }
+    fn compute_announced(&self, priority: u8) -> u64 {
+        self.port.compute_announced(priority)
+    }
     fn note_submission(&mut self) -> Result {
         self.effort.submit();
         super::Firmware::apply_shared_effort(self.init, self.effort, self.port, super::power::FULL_EFFORT)
@@ -1443,6 +1449,9 @@ impl compute::Host for DirectHost<'_> {
     fn compute_written(&self, priority: u8) -> u64 {
         self.port.compute_written(priority)
     }
+    fn compute_announced(&self, priority: u8) -> u64 {
+        self.port.compute_announced(priority)
+    }
     fn note_submission(&mut self) -> Result {
         Ok(())
     }
@@ -2373,6 +2382,12 @@ impl Registry {
         writeln!(f, "qos_ids used={} total=126", self.qos_ids.used())?;
         writeln!(f, "metrics_ids used={} total={}", self.metrics.used(), super::fw::buffer::METRICS_SLOTS)?;
         self.render.write_capacity(f)?;
+        let now = now_ns();
+        let waits = || self.compute.iter().flatten().filter_map(|entry| entry.queue.as_deref())
+            .filter_map(|queue| queue.kill_waiting(now));
+        compute_exit::write_kill_waits(f, waits().count(), waits().max().unwrap_or(0))?;
+        compute_exit::write_render_kills(f, self.teardown.kill_waits(now).count(),
+                                         self.teardown.kill_waits(now).max().unwrap_or(0))?;
         writeln!(f, "compute_slots used={} total={} reusable={} building={}",
                  self.compute.iter().flatten().count(), self.compute.len(),
                  self.compute.iter().flatten().filter_map(|entry| entry.queue.as_deref()).filter(|queue| queue.reusable()).count(),
diff --git a/drivers/gpu/drm/asahi/g17/runtime/backend.rs b/drivers/gpu/drm/asahi/g17/runtime/backend.rs
index d19f51c08dd290..bf0f989f042a50 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/backend.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/backend.rs
@@ -297,6 +297,16 @@ impl job::Backend for Backend {
         }
     }
 
+    fn long_job_beside(&self, _packet: &Arc<Packet>) -> bool {
+        let state = self.shared.state.lock();
+        let Some(firmware) = (*state).as_deref() else {
+            return false;
+        };
+        firmware.queues.compute.iter().flatten()
+            .filter_map(|entry| entry.queue.as_deref())
+            .any(|other| other.owner() != Some(self.owner) && other.long_job_running())
+    }
+
     fn defer_timeout(&self, packet: &Arc<Packet>, renewals: u8) -> bool {
         if packet.engine() == Engine::Compute {
             return self.defer_compute_timeout(packet, renewals);
diff --git a/drivers/gpu/drm/asahi/g17/runtime/compute_exit.rs b/drivers/gpu/drm/asahi/g17/runtime/compute_exit.rs
index bf9d023038a51b..9e12842f41c02c 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/compute_exit.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/compute_exit.rs
@@ -12,9 +12,117 @@ use crate::g17::{
     timeout::LIVENESS_NS,
     Firmware, MSG_CONTROL_NOTIFY,
 };
-use core::sync::atomic::{fence, Ordering};
+use core::sync::atomic::{fence, AtomicU32, AtomicU64, Ordering};
 use kernel::{prelude::*, sync::Arc};
 
+/// Waits of failed contexts' kills for a moment without other contexts' compute in flight,
+/// by bucket (under 10 ms, 100 ms, 1 s, 5 s, and longer) and the longest.
+static KILL_WAITS: [AtomicU32; 5] = [const { AtomicU32::new(0) }; 5];
+static KILL_WAIT_MAX_NS: AtomicU64 = AtomicU64::new(0);
+const KILL_WAIT_BOUNDS_NS: [u64; 4] = [10_000_000, 100_000_000, 1_000_000_000, 5_000_000_000];
+/// A failed context's kill waiting this long is reported, once per kill and at most once per
+/// this interval across kills.
+const SLOW_KILL_WAIT_NS: u64 = 5_000_000_000;
+static SLOW_KILL_WARNED_NS: AtomicU64 = AtomicU64::new(0);
+/// Failed contexts' kills published beside other contexts' compute under GPU memory pressure,
+/// and when the last one was: at most one per [`PRESSED_KILL_INTERVAL_NS`], so the memory of
+/// one is freed before another is considered.
+static PRESSED_KILLS: AtomicU32 = AtomicU32::new(0);
+static PRESSED_KILL_NS: AtomicU64 = AtomicU64::new(0);
+const PRESSED_KILL_INTERVAL_NS: u64 = 250_000_000;
+
+/// Render-only kills (see `Firmware::publish_render_kill`): published, acknowledged, and
+/// contexts refused (each counted once, by its first reason) because the scheduler page names
+/// a queue outside the context, because the held pairs have no single owner, or because
+/// admission could not hold them.
+static RENDER_KILLS: [AtomicU32; 5] = [const { AtomicU32::new(0) }; 5];
+const RENDER_KILL_PUBLISHED: usize = 0;
+const RENDER_KILL_ACKNOWLEDGED: usize = 1;
+const RENDER_KILL_SCOPE: usize = 2;
+const RENDER_KILL_OWNER: usize = 3;
+const RENDER_KILL_HOLD: usize = 4;
+/// A render-only candidate is left this long after close, as a quarantined compute queue is,
+/// so that ordinary retirement goes first.
+const RENDER_KILL_DELAY_NS: u64 = 250_000_000;
+
+/// The render queue a render-only kill names: one of the context's own render queues that
+/// its scheduler page lists, never the reserved queues 0 and 1.
+fn render_kill_target(listed: u128, own: u128) -> Option<u8> {
+    let candidates = listed & own & !0b11;
+    (candidates != 0).then(|| candidates.trailing_zeros() as u8)
+}
+
+/// A closed context may be killed on its render pairs' behalf once it has been closed for
+/// [`RENDER_KILL_DELAY_NS`], it was never killed, a pair of it holds executed but unstamped
+/// commands and no compute queue still runs it, switches away from it or killed it. A second kill
+/// of a killed context made the firmware fault on the killed pair's own tiling queue within a
+/// millisecond of acknowledging it: with the pair's release held back, a context killed through
+/// its compute queue still looked like a render-only candidate.
+fn render_kill_eligible(age: u64, killed: bool, executed: bool, compute_clear: bool) -> bool {
+    age >= RENDER_KILL_DELAY_NS && !killed && executed && compute_clear
+}
+
+/// The one admission owner of every pair a kill holds, if there is one.
+fn hold_owner(owners: impl Iterator<Item = Option<u64>>) -> Option<u64> {
+    let mut owner = None;
+    for slot_owner in owners {
+        match (owner, slot_owner) {
+            (_, None) | (_, Some(0)) => return None,
+            (None, Some(next)) => owner = Some(next),
+            (Some(current), Some(next)) if current != next => return None,
+            _ => {}
+        }
+    }
+    owner
+}
+
+fn refuse_render_kill(teardown: &mut crate::g17::teardown::Pending, index: usize, reason: usize) {
+    if teardown.refuse_render_kill(index) {
+        RENDER_KILLS[reason].fetch_add(1, Ordering::Relaxed);
+    }
+}
+
+/// GPU memory is within an eighth of the device limit. A failed context holds its memory
+/// until its kill is acknowledged, so kills waiting for other contexts' compute to pause can
+/// push the device to its limit, and then every client's allocations fail: on Neo,
+/// gnome-shell crashed on such an ENOMEM. A kill beside foreign compute risks one GPU
+/// recovery instead.
+fn memory_pressed() -> bool {
+    let ledger = &crate::memacct::LEDGER;
+    let limit = ledger.limit.load(Ordering::Relaxed);
+    limit != 0
+        && ledger
+            .user_committed
+            .load(Ordering::Relaxed)
+            .saturating_add(ledger.kernel_resident.load(Ordering::Relaxed))
+            > limit - limit / 8
+}
+
+fn record_kill_wait(waited: u64) {
+    let bucket = KILL_WAIT_BOUNDS_NS.iter().position(|bound| waited < *bound).unwrap_or(4);
+    KILL_WAITS[bucket].fetch_add(1, Ordering::Relaxed);
+    KILL_WAIT_MAX_NS.fetch_max(waited, Ordering::Relaxed);
+}
+
+pub(in crate::g17) fn write_kill_waits(f: &mut kernel::fmt::Formatter<'_>, waiting: usize, longest: u64)
+    -> kernel::fmt::Result
+{
+    let count = |bucket: usize| KILL_WAITS[bucket].load(Ordering::Relaxed);
+    writeln!(f, "failed_kill_waits under_10ms={} under_100ms={} under_1s={} under_5s={} longer={} max_ms={} memory_pressed={}",
+             count(0), count(1), count(2), count(3), count(4),
+             KILL_WAIT_MAX_NS.load(Ordering::Relaxed) / 1_000_000, PRESSED_KILLS.load(Ordering::Relaxed))?;
+    writeln!(f, "failed_kills waiting={} longest_wait_ms={}", waiting, longest / 1_000_000)
+}
+
+pub(in crate::g17) fn write_render_kills(f: &mut kernel::fmt::Formatter<'_>, waiting: usize, longest: u64)
+    -> kernel::fmt::Result
+{
+    let count = |index: usize| RENDER_KILLS[index].load(Ordering::Relaxed);
+    writeln!(f, "render_kills published={} acknowledged={} refused_scope={} refused_owner={} refused_hold={} waiting={} longest_wait_ms={}",
+             count(RENDER_KILL_PUBLISHED), count(RENDER_KILL_ACKNOWLEDGED), count(RENDER_KILL_SCOPE),
+             count(RENDER_KILL_OWNER), count(RENDER_KILL_HOLD), waiting, longest / 1_000_000)
+}
+
 impl super::Registry {
     /// ACK storage belongs to the already-retained transaction, so a full event
     /// batch cannot overflow a separate cache. No allocation or wait is needed.
@@ -26,6 +134,10 @@ impl super::Registry {
                 }
             }
         }
+        if self.teardown.acknowledge_render_kill(cookie) {
+            RENDER_KILLS[RENDER_KILL_ACKNOWLEDGED].fetch_add(1, Ordering::Relaxed);
+            crate::g17::trace::teardown(4, cookie, 0x100, 0);
+        }
     }
 }
 
@@ -53,16 +165,34 @@ impl Firmware {
                 let _ = queue.announce_exit(now, || self.primary.notify(MSG_CONTROL_NOTIFY));
             }
         }
+        let primary = &mut self.primary;
+        self.queues.teardown.announce_render_kills(|| primary.notify(MSG_CONTROL_NOTIFY));
+        // Under memory pressure one kill skips the wait, at most one per interval.
+        let last = PRESSED_KILL_NS.load(Ordering::Relaxed);
+        let mut pressed =
+            (last == 0 || now.saturating_sub(last) >= PRESSED_KILL_INTERVAL_NS) && memory_pressed();
         for index in 0..self.queues.teardown.len() {
             let Some((context, age)) = self.queues.teardown.closed(index, now) else {
                 continue;
             };
-            if !self.queues.render.teardown_idle(&context)?
-                || !self.queues.render.teardown_pools_released(&context)
-            {
+            if !self.queues.render.teardown_idle(&context)? {
+                self.queues.teardown.clear_kill_wait(index);
                 continue;
             }
-            if let Err(error) = self.publish_compute_exit(&context, age, now) {
+            // A failed context's executed pairs keep their pool leases until the kill is
+            // acknowledged (`Pair::firmware_drained`), so its kill cannot wait for them.
+            let result = if self.compute_failed(&context) || self.queues.render.teardown_pools_released(&context) {
+                // Only a render-only kill's wait lives on the entry; a compute kill keeps its
+                // own on the queue.
+                self.queues.teardown.clear_kill_wait(index);
+                self.publish_compute_exit(&context, age, now, &mut pressed)
+            } else if self.render_kill_candidate(index, &context, age) {
+                self.publish_render_kill(index, &context, now, &mut pressed)
+            } else {
+                self.queues.teardown.clear_kill_wait(index);
+                Ok(())
+            };
+            if let Err(error) = result {
                 dev_err!(
                     self.primary.state.shared.dev.as_ref(),
                     "Scheduler {} exit request failed: {:?}\n",
@@ -74,7 +204,7 @@ impl Firmware {
         Ok(())
     }
 
-    fn publish_compute_exit(&mut self, context: &Arc<Context>, age: u64, now: u64) -> Result {
+    fn publish_compute_exit(&mut self, context: &Arc<Context>, age: u64, now: u64, pressed: &mut bool) -> Result {
         let Some(index) = self.queues.compute.iter().position(|entry| {
             entry
                 .as_ref()
@@ -83,7 +213,36 @@ impl Firmware {
         }) else {
             return Ok(());
         };
-        if self.render_control_backpressured() || self.kill_unsafe(context, now) {
+        let failed = self.compute_failed(context);
+        if self.render_control_backpressured() {
+            return Ok(());
+        }
+        let beside = failed && self.foreign_compute_running(context, now);
+        if failed {
+            if beside && !*pressed {
+                let queue = self.queues.compute[index]
+                    .as_mut()
+                    .and_then(|entry| entry.queue.as_deref_mut());
+                if let Some(mut queue) = queue {
+                    let (waited, crossed) = queue.note_kill_wait(now, SLOW_KILL_WAIT_NS);
+                    let last = SLOW_KILL_WARNED_NS.load(Ordering::Relaxed);
+                    if crossed
+                        && (last == 0 || now.saturating_sub(last) >= SLOW_KILL_WAIT_NS)
+                        && SLOW_KILL_WARNED_NS
+                            .compare_exchange(last, now.max(1), Ordering::Relaxed, Ordering::Relaxed)
+                            .is_ok()
+                    {
+                        dev_warn!(
+                            self.primary.state.shared.dev.as_ref(),
+                            "Scheduler {} kill has waited {} ms for other contexts' compute to pause\n",
+                            context.id(),
+                            waited / 1_000_000
+                        );
+                    }
+                }
+                return Ok(());
+            }
+        } else if self.kill_unsafe(context, now) {
             return Ok(());
         }
         let (owner, target) = {
@@ -122,7 +281,7 @@ impl Firmware {
             && !self
                 .queues
                 .admission
-                .hold_render_slots_for_exit_kill(owner, hold)
+                .hold_render_slots_for_exit_kill(owner, hold, self.queues.render.executed_slots(context) & hold)
         {
             return Ok(());
         }
@@ -131,13 +290,25 @@ impl Firmware {
             self.queues.admission.release_exit_kill_hold(hold);
             return if error == EAGAIN { Ok(()) } else { Err(error) };
         }
-        // Publication is irrevocable even if its doorbell fails. Keep both
-        // the killed graph and the held render identities until processor stop.
+        // Publication is irrevocable even if its doorbell fails, so mark the context
+        // killed before anything below can fail. Keep the killed graph until processor
+        // stop, and the held render identities until this context's scheduler release
+        // is consumed or processors stop.
+        context.note_kill_published();
         let mut queue = self.queues.compute[index]
             .as_mut()
             .and_then(|entry| entry.queue.as_deref_mut())
             .ok_or(EIO)?;
+        if failed {
+            record_kill_wait(queue.take_kill_wait(now));
+        }
+        if beside {
+            *pressed = false;
+            PRESSED_KILLS.fetch_add(1, Ordering::Relaxed);
+            PRESSED_KILL_NS.store(now.max(1), Ordering::Relaxed);
+        }
         queue.record_exit(cookie, now);
+        crate::g17::trace::teardown(3, cookie, u64::from(context.id()), hold | u64::from(beside));
         fence(Ordering::SeqCst);
         if let Err(error) = queue.announce_exit(now, || self.primary.notify(MSG_CONTROL_NOTIFY)) {
             dev_warn!(
@@ -150,12 +321,141 @@ impl Firmware {
         Ok(())
     }
 
+    /// A closed context whose compute side is clean, with render pairs whose commands all ran
+    /// but were never stamped (`Pair::executed_for_kill`). Its failed compute command was
+    /// retired on the firmware's fault record, so no compute queue is left to carry a kill,
+    /// yet those pairs are released only after one: without this the context, its pairs and
+    /// its memory stay until the processors stop.
+    fn render_kill_candidate(&self, index: usize, context: &Arc<Context>, age: u64) -> bool {
+        render_kill_eligible(
+            age,
+            self.queues.teardown.render_kill_published(index) || context.kill_published(),
+            self.queues.render.executed_slots(context) != 0,
+            self.queues.compute.iter().flatten().filter_map(|entry| entry.queue.as_deref()).all(|queue| {
+                queue.idle_for_context(context) && !queue.switching_from(context) && !queue.exit_published(context)
+            }),
+        )
+    }
+
+    /// Kill a render-only candidate (`render_kill_candidate`). The rules of a failed
+    /// context's kill apply: the pairs are held for the kill and their memory is released only
+    /// once the scheduler page reports it, the kill waits for a moment without other contexts'
+    /// compute unless GPU memory is pressed, and it names only queues of this context.
+    fn publish_render_kill(&mut self, index: usize, context: &Arc<Context>, now: u64, pressed: &mut bool) -> Result {
+        if self.render_control_backpressured() {
+            return Ok(());
+        }
+        let beside = self.foreign_compute_running(context, now);
+        if beside && !*pressed {
+            let (waited, crossed) = self.queues.teardown.note_kill_wait(index, now, SLOW_KILL_WAIT_NS);
+            let last = SLOW_KILL_WARNED_NS.load(Ordering::Relaxed);
+            if crossed
+                && (last == 0 || now.saturating_sub(last) >= SLOW_KILL_WAIT_NS)
+                && SLOW_KILL_WARNED_NS
+                    .compare_exchange(last, now.max(1), Ordering::Relaxed, Ordering::Relaxed)
+                    .is_ok()
+            {
+                dev_warn!(
+                    self.primary.state.shared.dev.as_ref(),
+                    "Scheduler {} kill has waited {} ms for other contexts' compute to pause\n",
+                    context.id(),
+                    waited / 1_000_000
+                );
+            }
+            return Ok(());
+        }
+        let (mut own, mut foreign, slots) = self.queues.render.exit_scope(context);
+        for entry in self.queues.compute.iter().flatten() {
+            if let Some(queue) = entry.queue.as_deref() {
+                let bit = 1u128 << queue.qid();
+                if queue.context().is_some_and(|bound| Arc::ptr_eq(bound, context)) {
+                    own |= bit;
+                } else {
+                    foreign |= bit;
+                }
+            }
+        }
+        let state = exit::State::read(context)?;
+        let Some(qids) = render_kill_target(state.qids, own).and_then(|target| state.scope(target, own, foreign)) else {
+            refuse_render_kill(&mut self.queues.teardown, index, RENDER_KILL_SCOPE);
+            return Ok(());
+        };
+        let held = |slot: u8| slots & (1u64 << slot) != 0 && self.queues.render.qids_for_slot(slot) & qids != 0;
+        let range = 0..crate::hw::t8140::queues::RENDER_SLOTS as u8;
+        let hold = range.clone().filter(|slot| held(*slot)).fold(0u64, |hold, slot| hold | 1u64 << slot);
+        let owner = hold_owner(range.filter(|slot| held(*slot)).map(|slot| self.queues.render.slot_owner(slot)));
+        let Some(owner) = owner.filter(|_| hold != 0) else {
+            refuse_render_kill(&mut self.queues.teardown, index, RENDER_KILL_OWNER);
+            return Ok(());
+        };
+        // As for a failed context's kill: the cookie first, so that running out of cookies
+        // leaves no admission hold behind.
+        let cookie = exit::next_cookie()?;
+        if !self
+            .queues
+            .admission
+            .hold_render_slots_for_exit_kill(owner, hold, self.queues.render.executed_slots(context) & hold)
+        {
+            refuse_render_kill(&mut self.queues.teardown, index, RENDER_KILL_HOLD);
+            return Ok(());
+        }
+        let record = ControlRecord::ContextKill(ContextKill::new(context.scheduler_va(), cookie));
+        if let Err(error) = Rings::publish_control(&self.init, &record) {
+            self.queues.admission.release_exit_kill_hold(hold);
+            return if error == EAGAIN { Ok(()) } else { Err(error) };
+        }
+        // Publication is irrevocable, so mark the context killed before anything below
+        // can fail. The held pairs stay held until this context's scheduler release is
+        // consumed (`release_killed_slots` at detach).
+        context.note_kill_published();
+        self.queues.teardown.record_render_kill(index, cookie, now)?;
+        record_kill_wait(self.queues.teardown.take_kill_wait(index, now));
+        if beside {
+            *pressed = false;
+            PRESSED_KILLS.fetch_add(1, Ordering::Relaxed);
+            PRESSED_KILL_NS.store(now.max(1), Ordering::Relaxed);
+        }
+        RENDER_KILLS[RENDER_KILL_PUBLISHED].fetch_add(1, Ordering::Relaxed);
+        crate::g17::trace::teardown(7, cookie, u64::from(context.id()), hold | u64::from(beside));
+        fence(Ordering::SeqCst);
+        let primary = &mut self.primary;
+        self.queues.teardown.announce_render_kills(|| primary.notify(MSG_CONTROL_NOTIFY));
+        Ok(())
+    }
+
+    /// Another context's compute command is in flight. On Neo, kills of failed contexts beside
+    /// other clients' compute occasionally made the firmware recover the GPU (twice in ten
+    /// minutes of a five-lane load with 67 faults), even with nothing freed before the kill's
+    /// acknowledgement; such a kill takes the next moment without foreign compute instead.
+    /// Render work is not waited for. A GPU that retired nothing for [`LIVENESS_NS`] has
+    /// stopped.
+    fn foreign_compute_running(&self, context: &Arc<Context>, now: u64) -> bool {
+        let stalled = crate::g17::recovery::since_retired(now).is_none_or(|since| since >= LIVENESS_NS);
+        !stalled
+            && self.queues.compute.iter().flatten().filter_map(|entry| entry.queue.as_deref()).any(|queue| {
+                queue.in_flight() && !queue.context().is_some_and(|bound| Arc::ptr_eq(bound, context))
+            })
+    }
+
+    /// A compute command of `context` failed and the firmware holds commands behind it that it
+    /// never runs. `publish_compute_exit` is only reached once `teardown_idle` found all of
+    /// the context's render pairs idle; the pairs behind such a failure become idle when
+    /// their commands have executed (`Pair::firmware_drained`), so by then nothing of the
+    /// context runs any more.
+    fn compute_failed(&self, context: &Arc<Context>) -> bool {
+        self.queues.compute.iter().flatten().filter_map(|entry| entry.queue.as_deref()).any(|queue| {
+            queue.context().is_some_and(|bound| Arc::ptr_eq(bound, context)) && queue.fault_pending()
+        })
+    }
+
     /// A kill while other contexts' commands run makes the firmware restart the
     /// GPU, and repeated ones crash it; on an otherwise idle GPU it is safe. A
     /// closed context's commands left alone retire like any others once the GPU
     /// reaches them, so its queues stay quarantined and retained until then. Only
     /// a GPU that retired nothing for [`LIVENESS_NS`] has stopped, and then the
-    /// kill proceeds.
+    /// kill proceeds. A context whose compute failed ([`Self::compute_failed`])
+    /// is not left alone, since its held commands never retire: its kill waits
+    /// only for other contexts' compute ([`Self::foreign_compute_running`]).
     fn kill_unsafe(&self, context: &Arc<Context>, now: u64) -> bool {
         let stalled = crate::g17::recovery::since_retired(now).is_none_or(|since| since >= LIVENESS_NS);
         if stalled {
@@ -184,6 +484,9 @@ impl Firmware {
         {
             return true;
         }
+        if self.queues.teardown.render_kill_polling(now, exit::POLL_NS) {
+            return true;
+        }
         (0..self.queues.teardown.len()).any(|index| {
             self.queues
                 .teardown
@@ -201,3 +504,36 @@ impl Firmware {
         })
     }
 }
+
+#[kernel::macros::kunit_tests(asahi_g17_render_kill_target)]
+mod render_kill_target_tests {
+    use super::{hold_owner, render_kill_eligible, render_kill_target, RENDER_KILL_DELAY_NS};
+
+    #[test]
+    fn a_killed_context_is_never_killed_again() {
+        assert!(render_kill_eligible(RENDER_KILL_DELAY_NS, false, true, true));
+        assert!(!render_kill_eligible(RENDER_KILL_DELAY_NS, true, true, true));
+        assert!(!render_kill_eligible(RENDER_KILL_DELAY_NS - 1, false, true, true));
+        assert!(!render_kill_eligible(RENDER_KILL_DELAY_NS, false, false, true));
+        assert!(!render_kill_eligible(RENDER_KILL_DELAY_NS, false, true, false));
+    }
+
+    #[test]
+    fn a_render_kill_holds_pairs_of_one_owner_only() {
+        assert_eq!(hold_owner([Some(5), Some(5)].into_iter()), Some(5));
+        assert_eq!(hold_owner([Some(5), Some(6)].into_iter()), None);
+        assert_eq!(hold_owner([Some(5), None].into_iter()), None);
+        assert_eq!(hold_owner([Some(0)].into_iter()), None);
+        assert_eq!(hold_owner(core::iter::empty()), None);
+    }
+
+    #[test]
+    fn a_render_kill_names_an_own_listed_queue_never_0_or_1() {
+        // Listed by the scheduler page but not the context's own: no target.
+        assert_eq!(render_kill_target(1 << 40, 1 << 41), None);
+        // The reserved queues are never named.
+        assert_eq!(render_kill_target(0b11, 0b11), None);
+        assert_eq!(render_kill_target(0b111 | 1 << 40, 0b111 | 1 << 40), Some(2));
+        assert_eq!(render_kill_target(1 << 40 | 1 << 90, 1 << 90), Some(90));
+    }
+}
diff --git a/drivers/gpu/drm/asahi/g17/runtime/events.rs b/drivers/gpu/drm/asahi/g17/runtime/events.rs
index 4cf2dd0fafb970..16bcb50523d656 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/events.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/events.rs
@@ -15,6 +15,9 @@ use kernel::{prelude::*, sync::Arc};
 
 /// Wake interval while a tiler heap may still shrink: the native check period.
 const SHRINK_WAKE_NS: u64 = 1_000_000_000;
+/// Longest interval between two event passes that run every housekeeping step while
+/// completion records alone keep the worker busy.
+const HOUSEKEEPING_NS: u64 = 1_000_000;
 
 struct Service<'a> {
     firmware: &'a mut Firmware,
@@ -72,6 +75,15 @@ impl event::Queues for Service<'_> {
     fn context_killed(&mut self, cookie: u64) {
         self.firmware.queues.acknowledge_compute_exit(cookie);
     }
+    fn compute_fault(&mut self, qid: u8, stamp: u32) {
+        for entry in self.firmware.queues.compute.iter_mut().flatten() {
+            if let Some(mut queue) = entry.queue.as_deref_mut() {
+                if queue.qid() == qid {
+                    queue.note_fault(stamp);
+                }
+            }
+        }
+    }
     fn flush_tvb_reply(&mut self) -> Result<bool> {
         self.firmware.flush_tvb_reply()
     }
@@ -378,6 +390,10 @@ impl crate::g17::Shared {
     }
 
     fn runtime_event_turn(self: &Arc<Self>, deferred: &mut DeferredBatch) -> Result {
+        // Taken before consuming records: a request made during this pass gets the next one.
+        let requested = self
+            .housekeeping
+            .swap(false, core::sync::atomic::Ordering::AcqRel);
         let (recover, masks) = {
             let mut state = self.state.lock();
             let Some(firmware) = (*state).as_deref_mut() else {
@@ -438,24 +454,42 @@ impl crate::g17::Shared {
             .any(|queue| queue.in_flight());
         let in_flight = compute || firmware.queues.render_in_flight();
         let idle = firmware.idle_check(now, in_flight)?;
-        let lingers = if firmware.recovery.pending() {
-            None
-        } else {
-            firmware.expire_render_lingers(now)?
-        };
-        firmware.reclaim_drained_renders(deferred)?;
-        firmware.service_compute_exit(deferred)?;
-        let pools = firmware.service_compute_pools(deferred, now)?;
+        // Expiries, drained pairs, exits, pool and heap trimming and teardown give back what
+        // retirement freed; a completion needs none of them to retire or to signal. While work
+        // is in flight and nothing waits for capacity, a pass for firmware records alone skips
+        // them. Any requested pass, recovery, teardown, a waiter, the last retirement and, at
+        // the latest, the first pass after HOUSEKEEPING_NS run them all.
+        let housekeeping = requested
+            || recover
+            || !in_flight
+            || firmware.recovery.pending()
+            || !firmware.queues.teardown.is_empty()
+            || self.channels.has_waiters()
+            || self.progress.has_waiters()
+            || self.queues.has_waiters()
+            || self.renders.has_waiters()
+            || now.saturating_sub(self.housekeeping_ns.load(core::sync::atomic::Ordering::Relaxed))
+                >= HOUSEKEEPING_NS;
+        let mut lingers = None;
+        let mut pools = None;
         let mut shrinking = false;
-        if !firmware.recovery.pending() {
-            firmware
-                .queues
-                .vacate_evicted_render_pools(&firmware.init, deferred)?;
-            shrinking = firmware
-                .queues
-                .shrink_render_heaps(&firmware.init, now, deferred);
-        }
-        {
+        if housekeeping {
+            self.housekeeping_ns
+                .store(now, core::sync::atomic::Ordering::Relaxed);
+            if !firmware.recovery.pending() {
+                lingers = firmware.expire_render_lingers(now)?;
+            }
+            firmware.reclaim_drained_renders(deferred)?;
+            firmware.service_compute_exit(deferred)?;
+            pools = firmware.service_compute_pools(deferred, now)?;
+            if !firmware.recovery.pending() {
+                firmware
+                    .queues
+                    .vacate_evicted_render_pools(&firmware.init, deferred)?;
+                shrinking = firmware
+                    .queues
+                    .shrink_render_heaps(&firmware.init, now, deferred);
+            }
             let mut reclaim = self.reclaim.lock();
             firmware.service_teardown(&mut reclaim)?;
             if !reclaim.is_empty() {
@@ -507,7 +541,7 @@ impl crate::g17::Shared {
         {
             // A full bounded pass must not depend on a fresh notification for
             // records that arrived while it was consuming the same ring.
-            self.queue_events();
+            self.queue_event_records();
         }
         let grow_ready = firmware.render_grow_ready();
         drop(state);
diff --git a/drivers/gpu/drm/asahi/g17/runtime/render.rs b/drivers/gpu/drm/asahi/g17/runtime/render.rs
index b5649f059f5d2c..4e7eaa2d8d2468 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/render.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/render.rs
@@ -76,6 +76,9 @@ pub(super) struct RenderState {
     pools: pools::Pools,
     grow: grow::Grow,
     retired: KVec<crate::g17::queue::render::RetiredBinding>,
+    /// Graphs of released contexts the firmware may still walk (see `detach_context`): kept
+    /// until the firmware stops, when this state is dropped.
+    kept_graphs: KVec<crate::g17::queue::render::RetiredBinding>,
     next_pm_generation: u64,
 }
 
@@ -102,6 +105,7 @@ impl RenderState {
             pools: pools::Pools::new()?,
             grow: grow::Grow::new()?,
             retired: KVec::with_capacity(128, GFP_KERNEL)?,
+            kept_graphs: KVec::new(),
             next_pm_generation: 1,
         })
     }
@@ -300,6 +304,7 @@ impl Registry {
                         if let Some(pair) = entry.pair.as_ref() {
                             self.buffers.unreserve_owner(pair.buffer_id())?;
                         }
+                        self.note_build_room();
                     }
                     return Ok(Some(UnusedRenderSpare {
                         _entry: entry,
@@ -357,11 +362,25 @@ impl Registry {
             .any(|(slot, entry)| entry.is_some() && self.admission.owns_slot(owner, slot))
     }
 
+    /// Tell admission whether another render pair can still be built.
+    pub(super) fn note_build_room(&self) {
+        self.admission.note_build_room(self.qids.render_pair_fits());
+    }
+
     pub(super) fn ensure_render_ids(&mut self, slot: u8, owner: u64) -> Result {
         if owner == 0 || slot == 0 || usize::from(slot) >= RENDER_SLOTS {
             return Err(EINVAL);
         }
-        if self.render.entries[usize::from(slot)].is_none() {
+        if self.render.entries[usize::from(slot)].is_some() {
+            return Ok(());
+        }
+        let built = self.build_render_ids(slot, owner);
+        self.note_build_room();
+        built
+    }
+
+    fn build_render_ids(&mut self, slot: u8, owner: u64) -> Result {
+        {
             let identity = self.identities.reserve(2)?.start;
             let tiling = self.qids.reserve(identity, DataMaster::Tiling, None)?;
             let fragment = match self.qids.reserve(identity + 1, DataMaster::Fragment, None) {
@@ -405,6 +424,13 @@ impl Registry {
     /// ensure_render_ids cancels a first ID if the second cannot be reserved;
     /// validation, identity overflow, and cancellation errors still propagate.
     fn prepare_render_ids(&mut self, slot: u8, owner: u64, optional: bool) -> Result<bool> {
+        // A full queue table refuses a sibling without scanning it on every preparation.
+        if optional
+            && self.render.entries.get(usize::from(slot)).is_some_and(Option::is_none)
+            && !self.qids.render_pair_fits()
+        {
+            return Ok(false);
+        }
         match self.ensure_render_ids(slot, owner) {
             Ok(()) => Ok(true),
             Err(error) if optional && error == ENOSPC => Ok(false),
@@ -1031,10 +1057,10 @@ impl Registry {
                     |pair| pair.published_in_flight(),
                 );
             active
-                && entry.reservation.ids.into_iter().all(|id| {
-                    let qid = usize::from(id.qid());
-                    entry.event_masks[qid / 64] & (1u64 << (qid % 64)) != 0
-                })
+                && crate::g17::queue::render::completion_noted(
+                    entry.reservation.ids.map(kick::Id::qid),
+                    entry.event_masks,
+                )
         })
     }
     pub(in crate::g17) fn defer_render_timeout(
@@ -1119,6 +1145,8 @@ impl Registry {
 impl RenderState {
     pub(super) fn write_capacity(&self, f: &mut kernel::fmt::Formatter<'_>) -> kernel::fmt::Result {
         writeln!(f, "retired_render_bindings used={} total={}", self.retired.len(), self.retired.capacity())?;
+        writeln!(f, "firmware_linked_render_graphs kept={} leaked={}", self.kept_graphs.len(),
+                 teardown::LEAKED_GRAPHS.load(core::sync::atomic::Ordering::Relaxed))?;
         self.pools.write_capacity(f)?;
         for (slot, entry) in self.entries.iter().enumerate() {
             if let Some(context) = entry.as_ref().and_then(|entry| entry.context.as_ref()) {
diff --git a/drivers/gpu/drm/asahi/g17/runtime/render/direct.rs b/drivers/gpu/drm/asahi/g17/runtime/render/direct.rs
index 1f42d1e6f404f9..36eae4816f9672 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/render/direct.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/render/direct.rs
@@ -220,7 +220,8 @@ impl crate::g17::Firmware {
     ) -> Result {
         self.queues.render.return_pair(slot, &mut loan.pair)?;
         if self.queues.render.take_missed_event(slot) {
-            self.primary.state.shared.queue_events();
+            // The record is retired by the next pass's observation, like any completion.
+            self.primary.state.shared.queue_event_records();
         }
         let entry = self.queues.render.entry_mut(slot)?;
         // A failure or timeout recorded against the lent pair is settled by the
diff --git a/drivers/gpu/drm/asahi/g17/runtime/render/exit.rs b/drivers/gpu/drm/asahi/g17/runtime/render/exit.rs
index c47110cb76fadf..6c58f728b3eb59 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/render/exit.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/render/exit.rs
@@ -26,6 +26,27 @@ impl RenderState {
         (own, foreign, slots)
     }
 
+    /// Slots of `context`'s pairs whose commands all ran but stay unretired until its kill.
+    pub(in crate::g17::runtime) fn executed_slots(&self, context: &Arc<Context>) -> u64 {
+        let mut slots = 0;
+        for (slot, entry) in self.entries.iter().enumerate() {
+            if entry.as_ref().and_then(|entry| entry.pair.as_ref())
+                .is_some_and(|pair| pair.executed_for_kill(context))
+            {
+                slots |= 1u64 << slot;
+            }
+        }
+        slots
+    }
+
+    /// The admission owner whose identities the pair in `slot` was built with.
+    pub(in crate::g17::runtime) fn slot_owner(&self, slot: u8) -> Option<u64> {
+        self.entries
+            .get(usize::from(slot))
+            .and_then(Option::as_ref)
+            .map(|entry| entry.reservation.owner)
+    }
+
     pub(in crate::g17::runtime) fn qids_for_slot(&self, slot: u8) -> u128 {
         self.entries
             .get(usize::from(slot))
diff --git a/drivers/gpu/drm/asahi/g17/runtime/render/observe.rs b/drivers/gpu/drm/asahi/g17/runtime/render/observe.rs
index 52bdd964ca4ae7..51682b43728c05 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/render/observe.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/render/observe.rs
@@ -1,21 +1,22 @@
 // SPDX-License-Identifier: GPL-2.0-only OR MIT
 
-//! Event-driven render snapshots outside the device mutex. The preparation
-//! epoch and exact pair owner survive every read, cancellation and retry.
+//! Event-driven render snapshots. A pair's completion memory is read and the
+//! pair retired in one device-mutex hold, inside the preparation epoch.
 
 use super::super::DeferredBatch;
-use crate::g17::queue::render::Pair;
 use core::sync::atomic::Ordering;
 use kernel::{prelude::*, sync::Arc};
 
 impl crate::g17::Firmware {
-    fn take_render_observation(&mut self, start: usize) -> Result<Option<(usize, KBox<Pair>)>> {
+    /// The first slot from `start` whose pair has news to observe: published work in flight, the
+    /// fragment queue's notification seen (`completion_noted`), and an observation the pair accepts.
+    fn next_render_observation(&self, start: usize) -> Result<Option<usize>> {
         if self.recovery.pending() {
             return Ok(None);
         }
         let state = crate::g17::recovery::Memory::recovery_state(&self.init)?;
         for slot in start..self.queues.render.entries.len() {
-            let Some(entry) = self.queues.render.entries[slot].as_mut() else {
+            let Some(entry) = self.queues.render.entries[slot].as_ref() else {
                 continue;
             };
             if entry.preparing.is_some() || entry.growing {
@@ -26,9 +27,7 @@ impl crate::g17::Firmware {
             };
             let qids = pair.qids();
             if !pair.published_in_flight()
-                || !qids
-                    .into_iter()
-                    .all(|qid| entry.event_masks[usize::from(qid) / 64] & (1u64 << (qid % 64)) != 0)
+                || !crate::g17::queue::render::completion_noted(qids, entry.event_masks)
             {
                 continue;
             }
@@ -37,35 +36,13 @@ impl crate::g17::Firmware {
                 self.init.stamp(qids[1])?.load(Ordering::Relaxed),
             ];
             core::sync::atomic::fence(Ordering::Acquire);
-            let Some(packet) = pair.observation_packet(entry.event_masks, stamps, state)? else {
-                continue;
-            };
-            entry.borrowed_active = pair.published_in_flight();
-            entry.borrowed_dependencies = Some(pair.dependency_snapshot());
-            entry.borrowed_switch = pair.switch_in_flight();
-            entry.preparing = Some(packet);
-            return Ok(Some((slot, entry.pair.take().ok_or(EIO)?)));
+            if pair.observation_packet(entry.event_masks, stamps, state)?.is_some() {
+                return Ok(Some(slot));
+            }
         }
         Ok(None)
     }
 
-    fn return_render_observation(&mut self, slot: usize, pair: &mut Option<KBox<Pair>>) -> Result {
-        let returning = pair.as_ref().ok_or(EINVAL)?;
-        let entry = self.queues.render.entry_mut(slot as u8)?;
-        if entry.pair.is_some()
-            || entry.preparing.is_none()
-            || returning.slot() as usize != slot
-            || returning.qids() != entry.reservation.ids.map(crate::g17::kick::Id::qid)
-        {
-            return Err(EIO);
-        }
-        entry.pair = pair.take();
-        entry.borrowed_dependencies = None;
-        entry.borrowed_switch = None;
-        entry.preparing = None;
-        Ok(())
-    }
-
     pub(super) fn settle_render_terminal(
         &mut self,
         slot: usize,
@@ -122,57 +99,36 @@ impl crate::g17::Shared {
                 Err(error) if error == EAGAIN || error == ENODEV => break,
                 result => result?,
             };
-            let loan = {
-                let mut state = self.state.lock();
-                let Some(firmware) = (*state).as_deref_mut() else {
-                    return Ok(());
-                };
-                if !epoch.is_current() || self.crashed.load(Ordering::Acquire) {
-                    break;
-                }
-                firmware.take_render_observation(cursor)?
+            // Selecting, reading and retiring the front share one hold: a
+            // separate hold for the read would let publishers and the
+            // retirement contend twice per observed pair.
+            let mut state = self.state.lock();
+            let Some(firmware) = (*state).as_deref_mut() else {
+                return Ok(());
             };
-            let Some((slot, pair)) = loan else {
+            if !epoch.is_current()
+                || firmware.recovery.pending()
+                || self.crashed.load(Ordering::Acquire)
+            {
+                break;
+            }
+            let Some(slot) = firmware.next_render_observation(cursor)? else {
                 break;
             };
             if slot != cursor {
                 remaining = crate::hw::t8140::queues::RENDER_DEPTH;
                 retries = 0;
             }
-            let observed = pair.observe_front();
-            let mut pair = Some(pair);
-            let mut state = self.state.lock();
-            // Return installed ownership before examining read errors or epoch.
-            let restored = (*state)
-                .as_deref_mut()
-                .ok_or(ENODEV)
-                .and_then(|firmware| firmware.return_render_observation(slot, &mut pair));
-            if let Err(error) = restored {
-                // A failed invariant must not drop firmware-owned storage. Join
-                // the other preparers only after releasing our own epoch, and
-                // keep this detached owner until both processors have stopped.
-                drop(state);
-                drop(epoch);
-                self.lose_device_quietly();
-                self.stop_runtime();
-                if let Some(mut pair) = pair {
-                    pair.stopped(&mut |event| {
-                        event.finish();
-                        Ok(())
-                    })?;
-                }
-                return Err(error);
-            }
-            let firmware = (*state).as_deref_mut().ok_or(EIO)?;
+            let observed = firmware.queues.render.entries[slot]
+                .as_ref()
+                .and_then(|entry| entry.pair.as_ref())
+                .ok_or(EIO)?
+                .observe_front();
+            // Publishers waiting for render capacity recheck once this hold ends.
             self.note_changed();
             if firmware.render_grow_pending() || firmware.queues.render.take_missed_event(slot as u8) {
-                self.queue_events();
-            }
-            if !epoch.is_current()
-                || firmware.recovery.pending()
-                || self.crashed.load(Ordering::Acquire)
-            {
-                break;
+                // Observation and the growth check run on every pass.
+                self.queue_event_records();
             }
             let result = (|| {
                 let state = crate::g17::recovery::Memory::recovery_state(&firmware.init)?;
diff --git a/drivers/gpu/drm/asahi/g17/runtime/render/submit.rs b/drivers/gpu/drm/asahi/g17/runtime/render/submit.rs
index 1ef6742b689a78..e6379178138d16 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/render/submit.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/render/submit.rs
@@ -131,7 +131,8 @@ impl Backend {
                         .reclaim_unused_render_spare(Some(self.owner), true)?;
                     let ready = firmware.queues.has_render_ids(self.owner);
                     drop(state);
-                    if reclaimed.is_none() && !ready {
+                    // A full queue table makes admission reuse a free killed pair instead.
+                    if reclaimed.is_none() && !ready && !pool.killed_pair_free() {
                         self.wait_render_capacity(&mut observed, started, sleep)?;
                     }
                     drop(reclaimed);
diff --git a/drivers/gpu/drm/asahi/g17/runtime/render/teardown.rs b/drivers/gpu/drm/asahi/g17/runtime/render/teardown.rs
index e54b1f4638f587..fffa6d28dd289c 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/render/teardown.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/render/teardown.rs
@@ -38,7 +38,7 @@ impl RenderState {
                 return Ok(false);
             }
             if let Some(pair) = entry.pair.as_ref() {
-                if !pair.idle_for_context(context)? {
+                if !pair.idle_for_context(context)? && !pair.executed_for_kill(context) {
                     return Ok(false);
                 }
             }
@@ -123,6 +123,7 @@ impl RenderState {
         if !out.has_room(count, 1) {
             return Err(EBUSY);
         }
+        let linked = context.firmware_may_link();
         for entry in self.entries.iter_mut().flatten() {
             if !entry
                 .context
@@ -140,7 +141,14 @@ impl RenderState {
             let (mut parked, retired) = KBox::into_inner(pair).park();
             // This park follows the consumed release of `context`.
             parked.release_consumed(context.scheduler_va());
-            *target = Some(retired);
+            if linked {
+                // Guard (kill before free): the firmware may still walk this context's work
+                // list into the graph, so it is not freed or recycled before the firmware stops.
+                kernel::pr_warn_once!("G17: keeping render graphs of an unkilled context that still links work\n");
+                keep_graph(&mut self.kept_graphs, retired);
+            } else {
+                *target = Some(retired);
+            }
             entry.parked = Some(parked);
             // Detached backing retains these references until the worker runs.
             entry.context = None;
@@ -155,8 +163,12 @@ impl RenderState {
             if let Some(id) = self.retired[index].buffer_id() {
                 buffers.unreserve_owner(id)?;
             }
-            let target = out.binding_slot()?;
-            *target = Some(self.retired.remove(index).map_err(|_| EIO)?);
+            let retired = self.retired.remove(index).map_err(|_| EIO)?;
+            if linked {
+                keep_graph(&mut self.kept_graphs, retired);
+            } else {
+                *out.binding_slot()? = Some(retired);
+            }
         }
         Ok(())
     }
@@ -181,3 +193,23 @@ impl Registry {
         Ok(())
     }
 }
+
+/// Graphs that could not be recorded in `kept_graphs` and were leaked instead, for debugfs.
+pub(super) static LEAKED_GRAPHS: core::sync::atomic::AtomicU32 = core::sync::atomic::AtomicU32::new(0);
+
+/// Keeps `graph` until the firmware stops; if even that cannot be recorded, it is leaked
+/// rather than freed.
+fn keep_graph(
+    kept: &mut KVec<crate::g17::queue::render::RetiredBinding>,
+    graph: crate::g17::queue::render::RetiredBinding,
+) {
+    if kept.reserve(1, GFP_KERNEL).is_ok() {
+        if let Err(error) = kept.push_within_capacity(graph) {
+            LEAKED_GRAPHS.fetch_add(1, core::sync::atomic::Ordering::Relaxed);
+            core::mem::forget(error.0);
+        }
+    } else {
+        LEAKED_GRAPHS.fetch_add(1, core::sync::atomic::Ordering::Relaxed);
+        core::mem::forget(graph);
+    }
+}
diff --git a/drivers/gpu/drm/asahi/g17/runtime/teardown.rs b/drivers/gpu/drm/asahi/g17/runtime/teardown.rs
index 16bd5828265a41..0222ae4c478c87 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/teardown.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/teardown.rs
@@ -100,6 +100,8 @@ impl ReclaimBatch {
 
 struct Service<'a> {
     render: &'a mut super::render::RenderState,
+    admission: &'a crate::g17::queue::admission::Pool,
+    objects: &'a Arc<crate::g17::object::Pool>,
     compute: &'a [Option<ComputeEntry>],
     buffers: &'a mut BufferIds,
     init: &'a InitData,
@@ -137,8 +139,15 @@ impl Host for Service<'_> {
         self.primary.notify(MSG_CONTROL_NOTIFY)
     }
     fn detach(&mut self, context: &Arc<Context>) -> Result {
+        let linked = context.retain_linked_work(self.objects);
+        let (_, _, slots) = self.render.exit_scope(context);
+        crate::g17::trace::teardown(5, u64::from(context.id()), slots, u64::from(linked));
         self.render
             .detach_context(context, self.buffers, self.reclaim)?;
+        // The firmware consumed this context's scheduler release after its kill (if any)
+        // was acknowledged and its pairs went idle: it no longer names these pairs, so
+        // their kill hold ends here rather than at the next processor stop.
+        self.admission.release_killed_slots(slots);
         // detach_context preflights this slot together with every graph slot.
         // Retain it before Pending removes its reference under the device lock.
         self.reclaim.context(context)
@@ -173,6 +182,7 @@ impl Firmware {
         self.queues.render.observe_teardown_releases(&self.init)?;
         let recovery = self.recovery.pending() || self.init.recovery_state()? != 0;
         let Registry {
+            admission,
             render,
             compute,
             buffers,
@@ -181,6 +191,8 @@ impl Firmware {
         } = &mut self.queues;
         let mut service = Service {
             render,
+            admission,
+            objects: &self.pool,
             compute,
             buffers,
             init: &self.init,
diff --git a/drivers/gpu/drm/asahi/g17/teardown.rs b/drivers/gpu/drm/asahi/g17/teardown.rs
index 53518a76e91a47..fa2a72a77f0609 100644
--- a/drivers/gpu/drm/asahi/g17/teardown.rs
+++ b/drivers/gpu/drm/asahi/g17/teardown.rs
@@ -24,6 +24,45 @@ struct Entry {
     phase: Phase,
     armed: u64,
     announce_pending: bool,
+    render_kill: Option<RenderKill>,
+    /// Since when, and whether reported, a render-only kill has waited for other contexts'
+    /// compute to pause.
+    kill_wait: Option<(u64, bool)>,
+    /// A render-only kill of this context was refused (counted once per context).
+    kill_refused: bool,
+}
+
+/// The kill of a closed context whose compute side is clean, published so that its render
+/// pairs' executed but unstamped commands can be released (see `Pair::firmware_drained`).
+/// The scheduler page's killed state, not this record, proves the kill to the pairs.
+#[derive(Clone, Copy)]
+struct RenderKill {
+    cookie: u64,
+    published: u64,
+    announced: bool,
+    acknowledged: bool,
+}
+
+impl RenderKill {
+    fn new(cookie: u64, now: u64) -> Self {
+        Self {
+            cookie,
+            published: now,
+            announced: false,
+            acknowledged: false,
+        }
+    }
+    /// Only this kill's exact cookie acknowledges it, once.
+    fn acknowledge(&mut self, cookie: u64) -> bool {
+        if cookie == 0 || cookie != self.cookie || self.acknowledged {
+            return false;
+        }
+        self.acknowledged = true;
+        true
+    }
+    fn awaiting(&self, now: u64, window: u64) -> bool {
+        !self.acknowledged && now.saturating_sub(self.published) < window
+    }
 }
 
 /// Publication and announcement errors retain the entry; no timeout frees
@@ -104,6 +143,9 @@ impl Pending {
             phase,
             armed: now,
             announce_pending: false,
+            render_kill: None,
+            kill_wait: None,
+            kill_refused: false,
         };
         self.entries.push(entry, GFP_KERNEL)?;
         Ok(())
@@ -233,6 +275,94 @@ impl Pending {
         Ok(None)
     }
 
+    /// A render-only kill of the closed context at `index` was published.
+    pub(crate) fn render_kill_published(&self, index: usize) -> bool {
+        self.entries.get(index).is_some_and(|entry| entry.render_kill.is_some())
+    }
+
+    /// Record a published render-only kill. Publication is irrevocable: the record stays
+    /// with the entry until the context's scheduler release is consumed.
+    pub(crate) fn record_render_kill(&mut self, index: usize, cookie: u64, now: u64) -> Result {
+        let entry = self.entries.get_mut(index).ok_or(EINVAL)?;
+        if entry.render_kill.is_some() {
+            return Err(EINVAL);
+        }
+        entry.render_kill = Some(RenderKill::new(cookie, now));
+        Ok(())
+    }
+
+    /// Ring the doorbell for published render-only kills not yet announced. A failed
+    /// doorbell is retried on a later pass and never duplicates the record.
+    pub(crate) fn announce_render_kills(&mut self, mut notify: impl FnMut() -> Result) {
+        for entry in self.entries.iter_mut() {
+            if let Some(kill) = entry.render_kill.as_mut() {
+                if !kill.announced && !kill.acknowledged && notify().is_ok() {
+                    kill.announced = true;
+                }
+            }
+        }
+    }
+
+    /// Only the exact cookie of an unacknowledged render-only kill is accepted.
+    pub(crate) fn acknowledge_render_kill(&mut self, cookie: u64) -> bool {
+        self.entries
+            .iter_mut()
+            .filter_map(|entry| entry.render_kill.as_mut())
+            .any(|kill| kill.acknowledge(cookie))
+    }
+
+    /// A render-only kill awaits its acknowledgement within the polling window.
+    pub(crate) fn render_kill_polling(&self, now: u64, window: u64) -> bool {
+        self.entries
+            .iter()
+            .any(|entry| entry.render_kill.is_some_and(|kill| kill.awaiting(now, window)))
+    }
+
+    /// The render-only kill of the entry at `index` waits; how long, and whether this is the
+    /// first time it crossed `slow`.
+    pub(crate) fn note_kill_wait(&mut self, index: usize, now: u64, slow: u64) -> (u64, bool) {
+        let Some(entry) = self.entries.get_mut(index) else {
+            return (0, false);
+        };
+        let (since, warned) = entry.kill_wait.get_or_insert((now, false));
+        let waited = now.saturating_sub(*since);
+        let crossed = waited >= slow && !*warned;
+        *warned |= crossed;
+        (waited, crossed)
+    }
+
+    /// The render-only kill was published: how long it waited, if it did.
+    pub(crate) fn take_kill_wait(&mut self, index: usize, now: u64) -> u64 {
+        self.entries
+            .get_mut(index)
+            .and_then(|entry| entry.kill_wait.take())
+            .map_or(0, |(since, _)| now.saturating_sub(since))
+    }
+
+    /// A render-only kill of the context at `index` was refused: it no longer waits, and this
+    /// returns whether it is the first refusal for this context.
+    pub(crate) fn refuse_render_kill(&mut self, index: usize) -> bool {
+        let Some(entry) = self.entries.get_mut(index) else {
+            return false;
+        };
+        entry.kill_wait = None;
+        !core::mem::replace(&mut entry.kill_refused, true)
+    }
+
+    /// The context at `index` is not waiting for a render-only kill (any more).
+    pub(crate) fn clear_kill_wait(&mut self, index: usize) {
+        if let Some(entry) = self.entries.get_mut(index) {
+            entry.kill_wait = None;
+        }
+    }
+
+    /// How long each waiting render-only kill has waited.
+    pub(crate) fn kill_waits(&self, now: u64) -> impl Iterator<Item = u64> + '_ {
+        self.entries
+            .iter()
+            .filter_map(move |entry| entry.kill_wait.map(|(since, _)| now.saturating_sub(since)))
+    }
+
     /// Poll for a bounded window at close and again at actual release publication.
     /// Failed announcements remain retryable on later service even after polling
     /// expires. Time passage never substitutes for a firmware-consumption witness.
@@ -242,3 +372,21 @@ impl Pending {
         })
     }
 }
+
+#[kernel::macros::kunit_tests(asahi_g17_render_kill)]
+mod render_kill_tests {
+    use super::RenderKill;
+
+    #[test]
+    fn only_the_exact_cookie_acknowledges_a_render_kill_once() {
+        let mut kill = RenderKill::new(7, 1_000);
+        assert!(kill.awaiting(1_000, 10));
+        assert!(!kill.acknowledge(0));
+        assert!(!kill.acknowledge(8));
+        assert!(kill.awaiting(1_009, 10));
+        assert!(!kill.awaiting(1_010, 10));
+        assert!(kill.acknowledge(7));
+        assert!(!kill.acknowledge(7));
+        assert!(!kill.awaiting(1_001, 10));
+    }
+}
diff --git a/drivers/gpu/drm/asahi/g17/timeout.rs b/drivers/gpu/drm/asahi/g17/timeout.rs
index 7b4ef4137b7958..b83ea5d07f406e 100644
--- a/drivers/gpu/drm/asahi/g17/timeout.rs
+++ b/drivers/gpu/drm/asahi/g17/timeout.rs
@@ -66,13 +66,18 @@ const SHARED_RENEWALS: u8 = 14;
 /// nothing for this long has stopped.
 pub(crate) const LIVENESS_NS: u64 = 5_000_000_000;
 
-/// Renews a published job's deadline while some command retired within
-/// [`LIVENESS_NS`] (`since_retired`), at most [`SHARED_RENEWALS`] times.
-pub(crate) fn renew_shared(published: bool, since_retired: Option<u64>, renewals: &mut u8) -> bool {
-    if !published
-        || !since_retired.is_some_and(|since| since < LIVENESS_NS)
-        || *renewals >= SHARED_RENEWALS
-    {
+/// Whether some command retired within [`LIVENESS_NS`] (`since_retired`).
+pub(crate) fn retiring(since_retired: Option<u64>) -> bool {
+    since_retired.is_some_and(|since| since < LIVENESS_NS)
+}
+
+/// Renews a published job's deadline while the GPU is `shared`, at most
+/// [`SHARED_RENEWALS`] times. It is shared while some command retires
+/// ([`retiring`]), or while a long job of another queue runs beside a long job:
+/// two ray traversal dispatches of about 6 s each took 10 to 17 s side by side, and
+/// neither retires anything until it ends.
+pub(crate) fn renew_shared(published: bool, shared: bool, renewals: &mut u8) -> bool {
+    if !published || !shared || *renewals >= SHARED_RENEWALS {
         return false;
     }
     *renewals += 1;
@@ -244,21 +249,34 @@ mod tests {
     fn a_live_gpu_renews_a_published_job_for_thirty_seconds() {
         let mut renewals = 0;
         for poll in 1..=14u8 {
-            assert!(renew_shared(true, Some(4_900_000_000), &mut renewals));
+            assert!(renew_shared(true, retiring(Some(4_900_000_000)), &mut renewals));
             assert_eq!(renewals, poll);
         }
-        assert!(!renew_shared(true, Some(0), &mut renewals));
+        assert!(!renew_shared(true, true, &mut renewals));
         assert_eq!(POLL_MS * (u32::from(renewals) + 1), 30_000);
     }
     #[test]
     fn a_stopped_gpu_or_unpublished_job_keeps_the_deadline() {
         let mut renewals = 0;
-        assert!(!renew_shared(true, Some(LIVENESS_NS), &mut renewals));
-        assert!(!renew_shared(true, None, &mut renewals));
-        assert!(!renew_shared(false, Some(0), &mut renewals));
+        assert!(!renew_shared(true, retiring(Some(LIVENESS_NS)), &mut renewals));
+        assert!(!renew_shared(true, retiring(None), &mut renewals));
+        assert!(!renew_shared(false, true, &mut renewals));
         assert_eq!(renewals, 0);
     }
     #[test]
+    fn a_long_job_beside_another_shares_the_gpu_without_retirements() {
+        // Two traversal dispatches side by side: nothing retires, yet both progress.
+        let mut renewals = 0;
+        let long_job_beside = true;
+        let shared = retiring(Some(LIVENESS_NS + 1)) || long_job_beside;
+        for poll in 1..=14u8 {
+            assert!(renew_shared(true, shared, &mut renewals));
+            assert_eq!(renewals, poll);
+        }
+        // The thirty-second bound still ends a pair that never finishes.
+        assert!(!renew_shared(true, true, &mut renewals));
+    }
+    #[test]
     fn exhausted_or_invalid_credit_never_wraps() {
         for value in 4..=u8::MAX {
             let mut renewals = value;
diff --git a/drivers/gpu/drm/asahi/g17/trace.c b/drivers/gpu/drm/asahi/g17/trace.c
index 23666be389b969..9e93291f5114ec 100644
--- a/drivers/gpu/drm/asahi/g17/trace.c
+++ b/drivers/gpu/drm/asahi/g17/trace.c
@@ -54,3 +54,9 @@ void asahi_g17_trace_event(u8 phase)
 	if (trace_asahi_g17_event_enabled())
 		trace_asahi_g17_event(phase, __arch_counter_get_cntvct());
 }
+
+void asahi_g17_trace_teardown(u8 kind, u64 a, u64 b, u64 c)
+{
+	if (trace_asahi_g17_teardown_enabled())
+		trace_asahi_g17_teardown(kind, a, b, c);
+}
diff --git a/drivers/gpu/drm/asahi/g17/trace.h b/drivers/gpu/drm/asahi/g17/trace.h
index 7e5a50e774b598..d1529a2775fcca 100644
--- a/drivers/gpu/drm/asahi/g17/trace.h
+++ b/drivers/gpu/drm/asahi/g17/trace.h
@@ -125,6 +125,25 @@ TRACE_EVENT(asahi_g17_message,
 		  __entry->counter)
 );
 
+/* Failed-context teardown and render slot reuse (kind: see g17/trace.rs teardown()). */
+TRACE_EVENT(asahi_g17_teardown,
+	TP_PROTO(u8 kind, u64 a, u64 b, u64 c),
+	TP_ARGS(kind, a, b, c),
+	TP_STRUCT__entry(
+		__field(u8, kind)
+		__field(u64, a)
+		__field(u64, b)
+		__field(u64, c)
+	),
+	TP_fast_assign(
+		__entry->kind = kind;
+		__entry->a = a;
+		__entry->b = b;
+		__entry->c = c;
+	),
+	TP_printk("kind=%u a=%#llx b=%#llx c=%#llx", __entry->kind, __entry->a, __entry->b, __entry->c)
+);
+
 TRACE_EVENT(asahi_g17_event,
 	TP_PROTO(u8 phase, u64 counter),
 	TP_ARGS(phase, counter),
diff --git a/drivers/gpu/drm/asahi/g17/trace.rs b/drivers/gpu/drm/asahi/g17/trace.rs
index b76b46425d2ce9..d5f61f50a93f68 100644
--- a/drivers/gpu/drm/asahi/g17/trace.rs
+++ b/drivers/gpu/drm/asahi/g17/trace.rs
@@ -16,6 +16,7 @@ unsafe extern "C" {
     fn asahi_g17_trace_doorbell_enabled() -> bool;
     fn asahi_g17_trace_doorbell(engine: u8, power: u32);
     fn asahi_g17_trace_event(phase: u8);
+    fn asahi_g17_trace_teardown(kind: u8, a: u64, b: u64, c: u64);
 }
 
 pub(super) const RENDER: u8 = 0;
@@ -74,6 +75,16 @@ pub(super) fn message(kind: u8, value: u64) {
     unsafe { asahi_g17_trace_message(kind, value) };
 }
 
+/// Failed-context teardown: 1 compute fault (qid, stamp), 2 unstamped render retire (slot,
+/// ordinal), 3 kill published (cookie, context, held slots | 1 if it skipped the wait for other
+/// contexts' compute under memory pressure; slot 0 is never held), 4 kill acknowledged (cookie,
+/// qid, or 0x100 for a render-only kill), 5 detach (context, slots, linked), 6 render slot to a
+/// new owner (slot, old, new | killed << 63), 7 render-only kill published (as 3).
+pub(crate) fn teardown(kind: u8, a: u64, b: u64, c: u64) {
+    // SAFETY: Plain values only.
+    unsafe { asahi_g17_trace_teardown(kind, a, b, c) };
+}
+
 /// Event worker pass boundaries: 0 begin, 1 fences signalled, 2 end.
 pub(super) fn event(phase: u8) {
     // SAFETY: Plain values only.
diff --git a/drivers/gpu/drm/asahi/g17/trace_api.h b/drivers/gpu/drm/asahi/g17/trace_api.h
index d09f529a163f0f..0049aa00796342 100644
--- a/drivers/gpu/drm/asahi/g17/trace_api.h
+++ b/drivers/gpu/drm/asahi/g17/trace_api.h
@@ -20,5 +20,6 @@ bool asahi_g17_trace_doorbell_enabled(void);
 void asahi_g17_trace_doorbell(u8 engine, u32 power);
 void asahi_g17_trace_message(u8 kind, u64 value);
 void asahi_g17_trace_event(u8 phase);
+void asahi_g17_trace_teardown(u8 kind, u64 a, u64 b, u64 c);
 
 #endif /* _ASAHI_G17_TRACE_API_H */
diff --git a/drivers/gpu/drm/asahi/hw/t8140.rs b/drivers/gpu/drm/asahi/hw/t8140.rs
index 75925713bf10bb..47dc4d583a5fcb 100644
--- a/drivers/gpu/drm/asahi/hw/t8140.rs
+++ b/drivers/gpu/drm/asahi/hw/t8140.rs
@@ -417,9 +417,10 @@ pub(crate) mod scheduling {
 pub(crate) mod queues {
     pub(crate) const RENDER_SLOTS: usize = 64;
     // One physical pair per logical owner leaves room for every execution
-    // context in the shared 128-QID namespace. Commands queue at RENDER_DEPTH.
+    // context in the shared 128-QID namespace. Commands queue at RENDER_DEPTH,
+    // as many as a pair has command records (the render memory's RECORDS).
     pub(crate) const RENDER_PAIRS_PER_OWNER: usize = 1;
-    pub(crate) const RENDER_DEPTH: u8 = 16;
+    pub(crate) const RENDER_DEPTH: u8 = 32;
     pub(crate) const COMPUTE_DEPTH: usize = 16;
 }