diff --git a/drivers/gpu/drm/asahi/driver.rs b/drivers/gpu/drm/asahi/driver.rs
index d0dfc840265da..240d67be314d4 100644
--- a/drivers/gpu/drm/asahi/driver.rs
+++ b/drivers/gpu/drm/asahi/driver.rs
@@ -341,6 +341,11 @@ impl AsahiPlatformDriver {
                 return Err(error);
             }
         };
+        if let Err(error) =
+            crate::sysfs_exports::register_idle(raw_dev, crate::g17::power::IDLE_HOLD_MS.as_ptr())
+        {
+            dev_warn!(pdev.as_ref(), "idle_off_delay_ms not registered: {:?}\n", error);
+        }
         Ok(AsahiPlatformDriverData {
             _device: reg.device().into(),
             _reg: reg,
diff --git a/drivers/gpu/drm/asahi/g17/completion.rs b/drivers/gpu/drm/asahi/g17/completion.rs
index 0fc340b29f5b5..36c46a3823bdf 100644
--- a/drivers/gpu/drm/asahi/g17/completion.rs
+++ b/drivers/gpu/drm/asahi/g17/completion.rs
@@ -242,6 +242,9 @@ impl Drop for PooledTimestamps {
     }
 }
 
+/// Republications of one command whose earlier publications firmware dropped unexecuted.
+pub(crate) const MAX_REPLAYS: u8 = 3;
+
 #[pin_data]
 pub(crate) struct Completion {
     status: Arc<VmStatus>,
@@ -385,7 +388,7 @@ impl Completion {
         let accepted = self
             .replays
             .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| {
-                (count < 3).then_some(count + 1)
+                (count < MAX_REPLAYS).then_some(count + 1)
             })
             .is_ok();
         if accepted {
@@ -464,6 +467,10 @@ impl Completion {
         self.class.load(Ordering::Acquire) == 1
     }
 
+    pub(crate) fn guilty(&self) -> bool {
+        self.class.load(Ordering::Acquire) == 2
+    }
+
     /// Installed before the first firmware-visible prefix. Cancellation takes
     /// the same lock after claiming the signal, so it cannot miss a late lease.
     pub(crate) fn attach_render_lease(
@@ -497,7 +504,7 @@ impl Completion {
     /// Called under the device mutex before deferring the error signal.
     pub(crate) fn quarantine_render(&self) {
         if let Some(lease) = self.render.lock().as_ref() {
-            lease.quarantine();
+            lease.quarantine(self.spared());
         }
     }
 
@@ -528,8 +535,11 @@ impl Completion {
             feed.note_render_pass(ticks[2], ticks[3]);
         }
         if let Err(error) = result {
+            // Spared work cancelled after another VM's recovery is not this client's device loss.
             if self.spared() {
-                self.status.report_failure(error);
+                if error != ECANCELED {
+                    self.status.report_failure(error);
+                }
             } else if error != ECANCELED {
                 self.status.record(error);
             }
diff --git a/drivers/gpu/drm/asahi/g17/job.rs b/drivers/gpu/drm/asahi/g17/job.rs
index 575a9a7adc976..3b47d361c2ef9 100644
--- a/drivers/gpu/drm/asahi/g17/job.rs
+++ b/drivers/gpu/drm/asahi/g17/job.rs
@@ -336,7 +336,7 @@ impl<B: Backend> sched::JobImpl for Job<B> {
         }
         let deferred = job.backend.defer_timeout(&job.packet, job.victim_renewals);
         if deferred {
-            job.victim_renewals += 1;
+            job.victim_renewals = job.victim_renewals.saturating_add(1);
         }
         deferred
     }
diff --git a/drivers/gpu/drm/asahi/g17/kick.rs b/drivers/gpu/drm/asahi/g17/kick.rs
index 7aaca95992f8d..d7eedc849aef1 100644
--- a/drivers/gpu/drm/asahi/g17/kick.rs
+++ b/drivers/gpu/drm/asahi/g17/kick.rs
@@ -63,7 +63,7 @@ struct Identity {
 pub(crate) struct QueueIds {
     entries: [Option<Identity>; QID_COUNT],
     initial_render: [Option<Id>; 2],
-    compute_publication: u64,
+    publication: u64,
     leased: usize,
 }
 
@@ -72,7 +72,7 @@ impl QueueIds {
         kernel::try_init!(Self {
             entries <- pin_init::init_array_from_fn(|_| None),
             initial_render: [None; 2],
-            compute_publication: 0,
+            publication: 0,
             leased: 0,
         })
     }
@@ -174,10 +174,11 @@ impl QueueIds {
         Ok(id)
     }
 
-    /// Ordered under the device mutex; no atomic operation on the submission path.
-    pub(crate) fn next_compute_publication(&mut self) -> Result<u64> {
-        self.compute_publication = self.compute_publication.checked_add(1).ok_or(EOVERFLOW)?;
-        Ok(self.compute_publication)
+    /// Ordered under the device mutex; no atomic operation on the submission path. Compute and
+    /// render publications share it, so a watchdog can order work across both engines.
+    pub(crate) fn next_publication(&mut self) -> Result<u64> {
+        self.publication = self.publication.checked_add(1).ok_or(EOVERFLOW)?;
+        Ok(self.publication)
     }
 
     /// Call after preflighting the complete installation, before any address
diff --git a/drivers/gpu/drm/asahi/g17/mod.rs b/drivers/gpu/drm/asahi/g17/mod.rs
index 953f57aaef7df..740f1a3ef877b 100644
--- a/drivers/gpu/drm/asahi/g17/mod.rs
+++ b/drivers/gpu/drm/asahi/g17/mod.rs
@@ -36,11 +36,12 @@ mod job;
 mod capacity;
 mod kick;
 mod object;
-mod power;
+pub(crate) mod power;
 mod preparation;
 mod qos;
 mod queue;
 mod recovery;
+mod replay;
 mod runtime;
 pub(crate) mod status;
 mod teardown;
@@ -477,6 +478,7 @@ struct Firmware {
     rings: channel::Rings,
     recovery: recovery::State,
     effort: power::Effort,
+    hold: power::Hold,
     registration_power_release_pending: bool,
     queues: runtime::Registry,
     pool: Arc<object::Pool>,
@@ -520,6 +522,11 @@ impl Firmware {
     }
 
     fn release_registration_power(&mut self) -> Result {
+        // An idle hold keeps the cores powered past registration; it releases them itself.
+        if self.hold.active() {
+            self.registration_power_release_pending = false;
+            return Ok(());
+        }
         power::release(self)
     }
 
@@ -549,14 +556,49 @@ impl Firmware {
     /// Returns the delay in ns for a follow-up idle check. Queue owners report only submitted,
     /// unretired work; quarantined work retaining mappings is excluded.
     fn idle_check(&mut self, now_ns: u64, in_flight: bool) -> Result<Option<u64>> {
-        match self.effort.idle(now_ns, in_flight) {
-            power::IdleAction::None => Ok(None),
+        let effort = match self.effort.idle(now_ns, in_flight) {
+            power::IdleAction::None => None,
             power::IdleAction::Clear => {
                 self.apply_effort(0)?;
-                Ok(None)
+                None
+            }
+            power::IdleAction::Recheck(delay) => Some(delay),
+        };
+        Ok(match (effort, self.idle_hold(now_ns, in_flight)) {
+            (Some(a), Some(b)) => Some(a.min(b)),
+            (a, b) => a.or(b),
+        })
+    }
+
+    /// Drives the idle hold (see power::Hold); returns the delay in ns for its next check. A
+    /// refused control (recovery in progress) is retried on a later check.
+    fn idle_hold(&mut self, now_ns: u64, in_flight: bool) -> Option<u64> {
+        const RETRY_NS: u64 = 1_000_000;
+        let hold_ms = power::IDLE_HOLD_MS.load(Ordering::Relaxed);
+        for _ in 0..2 {
+            match self.hold.step(now_ns, in_flight, self.effort.submissions(), hold_ms) {
+                power::HoldAction::None => return None,
+                power::HoldAction::Recheck(delay) => return Some(delay),
+                power::HoldAction::Assert => {
+                    if self.set_idle_control(false).is_err() {
+                        return None;
+                    }
+                    self.hold.asserted();
+                }
+                power::HoldAction::Release => {
+                    if self.set_idle_control(true).is_err() {
+                        return Some(RETRY_NS);
+                    }
+                    self.hold.released();
+                    return None;
+                }
             }
-            power::IdleAction::Recheck(delay) => Ok(Some(delay)),
         }
+        None
+    }
+
+    fn set_idle_control(&mut self, enabled: bool) -> Result<u32> {
+        power::Control::set_idle(self, enabled)
     }
 }
 
@@ -980,6 +1022,7 @@ impl Gpu {
                 rings: channel::Rings::new(),
                 recovery: recovery::State::new(),
                 effort: power::Effort::new(),
+                hold: power::Hold::new(),
                 registration_power_release_pending: false,
                 queues <- runtime::Registry::new(shared.clusters, shared.descriptor_flags),
                 pool: object::Pool::new()?,
diff --git a/drivers/gpu/drm/asahi/g17/power.rs b/drivers/gpu/drm/asahi/g17/power.rs
index 96b363bc27447..7fef15084c50a 100644
--- a/drivers/gpu/drm/asahi/g17/power.rs
+++ b/drivers/gpu/drm/asahi/g17/power.rs
@@ -3,9 +3,19 @@
 //! Work-based firmware performance effort. Idle power gating is managed separately by the
 //! device-control ring; this input never keeps an otherwise idle GPU powered.
 
+use core::sync::atomic::AtomicU32;
+
 pub(super) const FULL_EFFORT: u32 = 0x10000;
 const IDLE_HOLD_NS: u64 = 16_000_000;
 
+/// The firmware's idle power-off delay as measured on T6040/25G83: the cores power off 2 ms after
+/// the last job. Not measured on other firmware; at or below it the hold stays off.
+pub(crate) const FIRMWARE_IDLE_OFF_MS: u32 = 2;
+
+/// The `idle_off_delay_ms` sysfs knob. Above [`FIRMWARE_IDLE_OFF_MS`] the driver keeps the GPU
+/// cores powered until the GPU has been idle this long ([`Hold`]).
+pub(crate) static IDLE_HOLD_MS: AtomicU32 = AtomicU32::new(FIRMWARE_IDLE_OFF_MS);
+
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
 pub(super) enum IdleAction {
     None,
@@ -17,6 +27,7 @@ pub(super) struct Effort {
     streaming: bool,
     idle_since: Option<u64>,
     applied: u32,
+    submissions: u64,
 }
 
 impl Effort {
@@ -25,12 +36,19 @@ impl Effort {
             streaming: false,
             idle_since: None,
             applied: 0,
+            submissions: 0,
         }
     }
 
     pub(super) fn submit(&mut self) {
         self.streaming = true;
         self.idle_since = None;
+        self.submissions = self.submissions.wrapping_add(1);
+    }
+
+    /// Count of published submissions, for [`Hold::step`].
+    pub(super) fn submissions(&self) -> u64 {
+        self.submissions
     }
 
     pub(super) fn applied(&self) -> u32 {
@@ -62,6 +80,83 @@ impl Effort {
     }
 }
 
+#[derive(Copy, Clone, Debug, PartialEq, Eq)]
+pub(super) enum HoldAction {
+    None,
+    /// Send the keep-powered idle policy.
+    Assert,
+    /// Send the power-off idle policy again.
+    Release,
+    Recheck(u64),
+}
+
+/// Host-side idle hold. On T6040/25G83 the firmware powers the cores off 2 ms after the last job
+/// and reads that delay only at boot; every wake-up then costs about 1 ms. With a longer
+/// [`IDLE_HOLD_MS`] the host sends the keep-powered idle policy once work appears and the power-off
+/// policy once no work has been submitted or in flight for that long. Checked from the event
+/// worker, which runs on every completion, well inside the firmware's 2 ms.
+pub(super) struct Hold {
+    active: bool,
+    seen: u64,
+    since: Option<u64>,
+}
+
+impl Hold {
+    pub(super) const fn new() -> Self {
+        Self {
+            active: false,
+            seen: 0,
+            since: None,
+        }
+    }
+
+    pub(super) fn active(&self) -> bool {
+        self.active
+    }
+
+    pub(super) fn step(
+        &mut self,
+        now_ns: u64,
+        in_flight: bool,
+        submissions: u64,
+        hold_ms: u32,
+    ) -> HoldAction {
+        let enabled = hold_ms > FIRMWARE_IDLE_OFF_MS;
+        let fresh = submissions != self.seen;
+        self.seen = submissions;
+        if in_flight || fresh {
+            self.since = None;
+        }
+        if !self.active {
+            return if enabled && (in_flight || fresh) {
+                HoldAction::Assert
+            } else {
+                HoldAction::None
+            };
+        }
+        if in_flight {
+            return HoldAction::None;
+        }
+        let hold_ns = if enabled { u64::from(hold_ms) * 1_000_000 } else { 0 };
+        let since = *self.since.get_or_insert(now_ns.max(1));
+        let idle = now_ns.saturating_sub(since);
+        if idle >= hold_ns {
+            HoldAction::Release
+        } else {
+            HoldAction::Recheck(hold_ns - idle)
+        }
+    }
+
+    pub(super) fn asserted(&mut self) {
+        self.active = true;
+    }
+
+    pub(super) fn released(&mut self) {
+        self.active = false;
+        self.since = None;
+    }
+}
+
 /// Single-producer control path under the device mutex. Event draining must not publish another
 /// control between `set_idle` and the exact three-counter retirement witness.
 pub(super) trait Control {
@@ -137,3 +232,31 @@ pub(super) fn release<C: Control>(control: &mut C) -> kernel::error::Result {
     }
     result
 }
+
+#[cfg_attr(not(test), kernel::macros::kunit_tests(asahi_g17_idle_hold))]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn holds_until_idle_for_the_delay_then_releases() {
+        let mut hold = Hold::new();
+        assert_eq!(hold.step(0, true, 1, 10), HoldAction::Assert);
+        hold.asserted();
+        assert_eq!(hold.step(1_000_000, false, 1, 10), HoldAction::Recheck(10_000_000));
+        assert_eq!(hold.step(6_000_000, false, 1, 10), HoldAction::Recheck(5_000_000));
+        // A new submission restarts the idle period.
+        assert_eq!(hold.step(7_000_000, false, 2, 10), HoldAction::Recheck(10_000_000));
+        assert_eq!(hold.step(17_000_000, false, 2, 10), HoldAction::Release);
+        hold.released();
+        assert!(!hold.active());
+    }
+
+    #[test]
+    fn the_default_delay_never_holds_and_lowering_it_releases() {
+        let mut hold = Hold::new();
+        assert_eq!(hold.step(0, true, 1, FIRMWARE_IDLE_OFF_MS), HoldAction::None);
+        assert_eq!(hold.step(0, true, 2, 10), HoldAction::Assert);
+        hold.asserted();
+        assert_eq!(hold.step(1, false, 2, FIRMWARE_IDLE_OFF_MS), HoldAction::Release);
+    }
+}
diff --git a/drivers/gpu/drm/asahi/g17/queue/admission.rs b/drivers/gpu/drm/asahi/g17/queue/admission.rs
index 7fc4d79a5f95c..dc445b15f7c91 100644
--- a/drivers/gpu/drm/asahi/g17/queue/admission.rs
+++ b/drivers/gpu/drm/asahi/g17/queue/admission.rs
@@ -63,6 +63,9 @@ struct State {
     owners: [u64; cfg::RENDER_SLOTS],
     epochs: [u64; cfg::RENDER_SLOTS],
     quarantined: u64,
+    /// Quarantined slots whose failed work is all spared (innocent): waiting publications fail
+    /// their own fence with EBUSY instead of a device-loss error, so the VM stays usable.
+    spared: u64,
     kill_hold: u64,
     released: u64,
     epoch: u64,
@@ -78,6 +81,7 @@ impl State {
             owners: [0; cfg::RENDER_SLOTS],
             epochs: [0; cfg::RENDER_SLOTS],
             quarantined: 0,
+            spared: 0,
             kill_hold: 0,
             released: 0,
             epoch: 0,
@@ -425,7 +429,14 @@ impl Pool {
                 .is_some_and(|slot| state.quarantined & (1u64 << slot) != 0)
             {
                 if now >= quarantine_deadline {
-                    return Err(EIO);
+                    // Behind an innocent pair's quarantine (another VM wedged the GPU) only this
+                    // publication fails, not the VM: it recovers once the pair drains.
+                    let spared = state
+                        .owners
+                        .iter()
+                        .position(|id| *id == owner)
+                        .is_some_and(|slot| state.spared & (1u64 << slot) != 0);
+                    return Err(if spared { EBUSY } else { EIO });
                 }
                 let slice = remaining_ms.min(wait_slice_ms(quarantine_deadline, now)?);
                 if matches!(self.released.wait_interruptible_timeout(
@@ -556,6 +567,7 @@ impl Pool {
         }
         state.released &= !released;
         state.quarantined &= !released;
+        state.spared &= !released;
         state.kill_hold = 0;
         drop(state);
         self.released.notify_all();
@@ -661,14 +673,26 @@ impl RenderLease {
     pub(crate) fn slot(&self) -> u8 {
         self.slot
     }
-    pub(crate) fn quarantine(&self) {
-        self.pool.state.lock().quarantined |= 1u64 << self.slot;
+    /// `spared`: the failing work is innocent. Any guilty failure makes the slot's quarantine
+    /// guilty for good; only a quarantine that starts spared and stays spared is marked spared.
+    pub(crate) fn quarantine(&self, spared: bool) {
+        let bit = 1u64 << self.slot;
+        let mut state = self.pool.state.lock();
+        if !spared {
+            state.spared &= !bit;
+        } else if state.quarantined & bit == 0 {
+            state.spared |= bit;
+        }
+        state.quarantined |= bit;
+        drop(state);
         self.pool.released.notify_all();
     }
     /// The VM pin must be retired first. Keep this lease until after clearing
     /// quarantine; its drop is what permits waiting publications to proceed.
     pub(crate) fn rehabilitate_after_processor_stop(&self) {
-        self.pool.state.lock().quarantined &= !(1u64 << self.slot);
+        let mut state = self.pool.state.lock();
+        state.quarantined &= !(1u64 << self.slot);
+        state.spared &= !(1u64 << self.slot);
     }
     /// Requires the runtime's closed-recovery, exact pair-drain witness.
     pub(crate) fn rehabilitate_after_firmware_drain(&self) {
@@ -749,13 +773,35 @@ impl Pool {
 mod fault_capacity_tests {
     use super::*;
 
+    #[test]
+    fn a_guilty_failure_makes_a_spared_quarantine_guilty() {
+        let pool = Pool::new().unwrap();
+        let queue = pool.reserve_queue().unwrap();
+        let lease = pool.acquire_render(1, None, u64::MAX, || Ok(())).unwrap();
+        let bit = 1u64 << lease.slot();
+        lease.quarantine(true);
+        assert_ne!(pool.state.lock().spared & bit, 0);
+        lease.quarantine(false);
+        assert_eq!(pool.state.lock().spared & bit, 0);
+        // Once guilty, a later spared failure does not make it spared again.
+        lease.quarantine(true);
+        assert_eq!(pool.state.lock().spared & bit, 0);
+        assert_ne!(pool.state.lock().quarantined & bit, 0);
+        pool.release_owner(1);
+        lease.rehabilitate_after_firmware_drain();
+        drop(lease);
+        drop(queue);
+        let state = pool.state.lock();
+        assert_eq!(state.quarantined | state.spared, 0);
+    }
+
     #[test]
     fn closed_faulting_clients_return_leases_after_drain() {
         let pool = Pool::new().unwrap();
         for owner in 1..=512 {
             let queue = pool.reserve_queue().unwrap();
             let lease = pool.acquire_render(owner, None, u64::MAX, || Ok(())).unwrap();
-            lease.quarantine();
+            lease.quarantine(false);
             pool.release_owner(owner);
             let bit = 1u64 << lease.slot();
             assert_ne!(pool.state.lock().active & bit, 0);
diff --git a/drivers/gpu/drm/asahi/g17/queue/render.rs b/drivers/gpu/drm/asahi/g17/queue/render.rs
index b69cd4d565736..e1d2c64f6c349 100644
--- a/drivers/gpu/drm/asahi/g17/queue/render.rs
+++ b/drivers/gpu/drm/asahi/g17/queue/render.rs
@@ -87,6 +87,8 @@ pub(crate) trait Host {
     fn qos_cancel(&mut self, publication: qos::Publication) -> Result;
     fn install_pair(&mut self, registration: &RenderRegistration) -> Result;
     fn publish_qid(&mut self, id: kick::Id) -> Result;
+    /// The next publication number, shared with compute (timeout ordering).
+    fn next_publication(&mut self) -> Result<u64>;
     fn publish_outer_pair(
         &mut self,
         priority: u8,
@@ -120,6 +122,8 @@ struct Prepared {
     mcache: Option<McacheTable>,
 }
 struct Active {
+    /// Publication number, shared with compute (crate::g17::timeout).
+    published: u64,
     qos_completed: bool,
     flist_completed: bool,
     outer_published: bool,
@@ -169,8 +173,10 @@ pub(crate) struct Pair {
     previous: Option<RetiredBinding>,
     displaced: Option<RetiredBinding>,
     context_update: bool,
-    /// A retired, never-executed group awaiting republication on this pair.
-    replay: Option<Arc<Packet>>,
+    /// Retired, never-executed groups awaiting republication on this pair, in
+    /// their original order. Nothing else is prepared on the pair before them.
+    replay: Tracker<Arc<Packet>>,
+    /// When the oldest parked group was parked or the last one republished.
     replay_since: Option<u64>,
     /// Scheduler page of the context of the last published command: the
     /// scheduler the firmware last ran these queues for. Zero for a fresh graph.
@@ -266,7 +272,7 @@ impl Pair {
             terminal: None,
             previous: None,
             displaced: None,
-            replay: None,
+            replay: Tracker::new(),
             replay_since: None,
             context_update: true,
             executed: 0,
@@ -298,24 +304,25 @@ impl Pair {
         &mut self.memory.manager
     }
     pub(crate) fn in_flight(&self) -> bool {
-        self.active.len() != 0 || self.replay.is_some()
+        self.active.len() != 0 || self.replay.len() != 0
     }
     /// A replay or an abandonment grace period needs the worker to poll.
     pub(crate) fn replay_polling(&self) -> bool {
-        self.replay.is_some() || self.active.iter().any(|a| a.abandoned_since.is_some())
+        self.replay.len() != 0 || self.active.iter().any(|a| a.abandoned_since.is_some())
     }
     pub(crate) fn replay_stalled(&self, now: u64) -> bool {
-        self.replay.is_some() && crate::g17::timeout::background_stalled(self.replay_since, now)
+        self.replay.len() != 0 && crate::g17::timeout::background_stalled(self.replay_since, now)
     }
     pub(crate) fn replay_front(&self) -> Option<Arc<Packet>> {
-        self.replay.clone()
+        self.replay.front().cloned()
     }
     pub(crate) fn finish_replay(&mut self, packet: &Arc<Packet>) -> Result {
-        if !self.replay.as_ref().is_some_and(|front| Arc::ptr_eq(front, packet)) {
+        if !self.replay.front().is_some_and(|front| Arc::ptr_eq(front, packet)) {
             return Err(EIO);
         }
-        self.replay = None;
-        self.replay_since = None;
+        self.replay.pop();
+        // Progress: the expiry restarts for the next parked group, if any.
+        self.replay_since = (self.replay.len() != 0).then(crate::g17::now_ns);
         Ok(())
     }
     pub(crate) fn published_in_flight(&self) -> bool {
@@ -337,6 +344,7 @@ impl Pair {
         addresses: RenderAddresses,
     ) -> Result {
         if !self.room()
+            || self.replay.front().is_some_and(|front| !Arc::ptr_eq(front, packet))
             || !Arc::ptr_eq(&self.memory.context, &packet.context)
             || !packet.context.is_current()
         {
@@ -724,8 +732,10 @@ impl Pair {
         let next_inner = (next_ordinal as u32).wrapping_mul(2);
         let flist_submission = self.free_list.prepare_submission(2)?;
         let pm_submission = memory.manager.prepare_submission()?;
+        let published = host.next_publication()?;
         self.active
             .push(Active {
+                published,
                 qos_completed: false,
                 flist_completed: false,
                 outer_published: false,
@@ -956,6 +966,21 @@ impl Pair {
                 .all(|qid| masks[usize::from(qid) / 64] & (1u64 << (qid % 64)) != 0)
         })
     }
+    /// The oldest published group, for cross-VM timeout attribution (crate::g17::timeout).
+    /// The oldest published group still running. A quarantined pair's work holds the GPU until
+    /// the firmware drains it, so it stays visible to other VMs' watchdogs, as compute does.
+    pub(crate) fn timeout_head(&self) -> Option<crate::g17::timeout::Head> {
+        self.active
+            .iter()
+            .filter(|active| active.outer_published)
+            .find(|active| active.drained_result.is_none())
+            .map(|active| crate::g17::timeout::Head {
+                vm: Arc::as_ptr(active.packet.completion.status()) as usize,
+                published: active.published,
+                finished: false,
+            })
+    }
+
     pub(crate) fn contains_packet(&self, packet: &Arc<Packet>) -> bool {
         self.active
             .iter()
@@ -966,7 +991,7 @@ impl Pair {
                 .is_some_and(|prepared| Arc::ptr_eq(&prepared.packet, packet))
     }
     pub(crate) fn replay_contains(&self, packet: &Arc<Packet>) -> bool {
-        self.replay.as_ref().is_some_and(|replay| Arc::ptr_eq(replay, packet))
+        self.replay.iter().any(|replay| Arc::ptr_eq(replay, packet))
     }
     pub(crate) fn take_deferred_timeout(&self) -> Option<(Arc<Packet>, Error)> {
         self.active.iter().find_map(|active| {
@@ -1142,7 +1167,7 @@ impl Pair {
         };
         let timestamps = match result {
             Ok(retirement::Poll::Complete(timestamps)) => timestamps,
-            Ok(retirement::Poll::Pending) => return self.replay_abandoned(host, defer),
+            Ok(retirement::Poll::Pending) => return self.replay_abandoned(host, defer, true),
             Ok(retirement::Poll::Retry) => return Ok(false),
             Err(error) => {
                 self.quarantine(error, defer)?;
@@ -1172,105 +1197,144 @@ impl Pair {
         Ok(true)
     }
 
-    /// The front group was retired and unlinked by firmware without executing
-    /// (job list empty, stamps cover it, no recovery, fragment end never
-    /// written). Nothing of this group reached its attachments, so the same
-    /// command is published again instead of failing the owner's VM. The
-    /// proof is the strict timeout witness, taken twice ABANDON_GRACE_NS apart,
-    /// and only for a lone group whose completion still accepts a replay.
+    /// Every retained group was retired and unlinked by firmware without
+    /// executing (job list empty, stamps cover it, no recovery, fragment end
+    /// never written): after another VM's recovery the 25G83 firmware drops an
+    /// innocent pair's queued groups this way. Nothing of them reached their
+    /// attachments, so the same commands are published again, in order, instead
+    /// of failing their owners. Only the whole set is replayed; a replayed prefix
+    /// would run after a live successor. `grace` asks for the strict witness
+    /// twice ABANDON_GRACE_NS apart (worker polling); the render timeout, long
+    /// after publication, takes it once as its retirement does.
     fn replay_abandoned(
         &mut self,
         host: &mut impl Host,
         defer: &mut impl FnMut(Deferred) -> Result,
+        grace: bool,
     ) -> Result<bool> {
         use retirement::Memory;
-        if self.quarantined || self.active.len() != 1 || self.replay.is_some() {
+        if self.quarantined || self.active.len() == 0 || self.replay.len() != 0 {
             return Ok(false);
         }
-        let front = self.active.front().ok_or(EIO)?;
-        if !front.outer_published || front.failed {
-            return Ok(false);
+        use crate::g17::replay::{self, Decision, Group};
+        let qids = self.qids();
+        let memory = &self.memory;
+        let mut failure = None;
+        let groups = self.active.iter().map(|active| {
+            let innocent =
+                active.outer_published && !active.failed && !active.packet.completion.guilty();
+            let exhausted =
+                active.packet.completion.replay_count() >= crate::g17::completion::MAX_REPLAYS;
+            // decide() stops at the first group that is not abandoned, an error included.
+            let abandoned = innocent
+                && (|| -> Result<bool> {
+                    let mut observer = Observer {
+                        memory,
+                        qids,
+                        host: &mut *host,
+                    };
+                    core::sync::atomic::fence(core::sync::atomic::Ordering::Acquire);
+                    let snapshot = observer.snapshot(active.ticket.ordinal)?;
+                    let stamps = observer.stamps()?;
+                    let (_, state) = observer.host.epoch()?;
+                    // A torn sample is no proof; a later poll or the timeout looks again.
+                    Ok(matches!(
+                        retirement::abandoned(&active.ticket, snapshot, stamps, state, false),
+                        Ok(true)
+                    ))
+                })()
+                .unwrap_or_else(|error: Error| {
+                    failure = Some(error);
+                    false
+                });
+            Group {
+                innocent,
+                abandoned,
+                exhausted,
+            }
+        });
+        let decision = replay::decide(groups);
+        if let Some(error) = failure {
+            return Err(error);
         }
-        let mut observer = Observer {
-            memory: &self.memory,
-            qids: self.qids(),
-            host,
-        };
-        core::sync::atomic::fence(core::sync::atomic::Ordering::Acquire);
-        let snapshot = observer.snapshot(front.ticket.ordinal)?;
-        let stamps = observer.stamps()?;
-        let (_, state) = observer.host.epoch()?;
-        let abandoned =
-            retirement::abandoned(&front.ticket, snapshot, stamps, state, false)?;
         let now = crate::g17::now_ns();
         let front = self.active.front_mut().ok_or(EIO)?;
-        if !abandoned {
-            front.abandoned_since = None;
-            return Ok(false);
+        match decision {
+            Decision::Refuse => return Ok(false),
+            Decision::Wait => {
+                front.abandoned_since = None;
+                return Ok(false);
+            }
+            Decision::Replay => {}
         }
-        let Some(since) = front.abandoned_since else {
-            front.abandoned_since = Some(now);
-            return Ok(false);
-        };
-        if now.saturating_sub(since) < ABANDON_GRACE_NS {
-            return Ok(false);
+        if grace {
+            let Some(since) = front.abandoned_since else {
+                front.abandoned_since = Some(now);
+                return Ok(false);
+            };
+            if now.saturating_sub(since) < ABANDON_GRACE_NS {
+                return Ok(false);
+            }
         }
-        if crate::debug::debug_enabled(crate::debug::DebugFlags::VerboseFaults) {
-            pr_info!(
-                "render slot {} retired without execution (pass {:?}); replaying\n",
+        if !grace || crate::debug::debug_enabled(crate::debug::DebugFlags::VerboseFaults) {
+            pr_warn!(
+                "G17: render slot {}: {} groups retired without execution; replaying\n",
                 self.slot,
-                snapshot.pass
+                self.active.len()
             );
         }
         let qos = self.qos_owners();
         let buffer = self.buffer_id();
         let pool = self.free_list.id();
-        let front = self.active.front_mut().ok_or(EIO)?;
-        if !front.flist_completed {
-            host.complete_render(
-                (!front.qos_completed).then_some(qos),
-                pool,
-                front.generation,
-                buffer,
-            )?;
-            front.flist_completed = true;
-            front.qos_completed = true;
-        }
-        let ordinal = front.ticket.ordinal;
-        let packet = front.packet.clone();
-        if !packet.completion.begin_replay() {
-            if !packet.completion.spared() {
-                packet.completion.status().record(EIO);
-                if self.terminal.is_none() {
-                    self.terminal = Some((packet.completion.status().clone(), EIO));
+        // Every group parked below shares this expiry (an early return keeps it).
+        self.replay_since = Some(now);
+        while let Some(front) = self.active.front_mut() {
+            if !front.flist_completed {
+                host.complete_render(
+                    (!front.qos_completed).then_some(qos),
+                    pool,
+                    front.generation,
+                    buffer,
+                )?;
+                front.flist_completed = true;
+                front.qos_completed = true;
+            }
+            let ordinal = front.ticket.ordinal;
+            let packet = front.packet.clone();
+            if !packet.completion.begin_replay() {
+                if !packet.completion.spared() {
+                    packet.completion.status().record(EIO);
+                    if self.terminal.is_none() {
+                        self.terminal = Some((packet.completion.status().clone(), EIO));
+                    }
                 }
+                defer(Deferred::Retired(packet.completion.clone(), Err(EIO)))?;
+                self.memory.retire(ordinal)?;
+                self.active.pop().ok_or(EIO)?;
+                return Ok(true);
             }
-            defer(Deferred::Retired(packet.completion.clone(), Err(EIO)))?;
             self.memory.retire(ordinal)?;
             self.active.pop().ok_or(EIO)?;
-            return Ok(true);
+            packet.reset_publication_for_replay();
+            self.replay.push(packet).map_err(|_| EIO)?;
         }
-        self.memory.retire(ordinal)?;
-        self.active.pop().ok_or(EIO)?;
-        packet.reset_publication_for_replay();
-        self.replay = Some(packet);
-        self.replay_since = Some(now);
         Ok(true)
     }
 
     /// The pending replay cannot proceed (closed owner, cancellation or a
-    /// publication refusal). ECANCELED leaves the VM usable; any other error
-    /// is the ordinary device-loss outcome the replay tried to avoid.
+    /// publication refusal). Its successors cannot run without it and fail with
+    /// it. ECANCELED leaves the VM usable; any other error is the ordinary
+    /// device-loss outcome the replay tried to avoid.
     pub(crate) fn fail_replay(
         &mut self,
         error: Error,
         defer: &mut impl FnMut(Deferred) -> Result,
     ) -> Result {
-        let Some(packet) = self.replay.take() else {
-            return Ok(());
-        };
         self.replay_since = None;
-        defer(Deferred::Retired(packet.completion.clone(), Err(error)))
+        while let Some(packet) = self.replay.pop() {
+            defer(Deferred::Retired(packet.completion.clone(), Err(error)))?;
+        }
+        Ok(())
     }
 
     /// A closed preparation epoch invalidates only this unpublished descriptor.
@@ -1389,22 +1453,45 @@ impl Pair {
         if !self.active.iter().any(|a| Arc::ptr_eq(&a.packet, packet)) {
             return Ok(false);
         }
-        if !self.quarantined && self.active.len() == 1 {
-            let active = self.active.front().ok_or(EIO)?;
-            let mut observer = Observer {
-                memory: &self.memory,
-                qids: self.qids(),
-                host,
-            };
-            core::sync::atomic::fence(core::sync::atomic::Ordering::Acquire);
-            let snapshot = observer.snapshot(active.ticket.ordinal)?;
-            let stamps = observer.stamps()?;
-            let (_, state) = observer.host.epoch()?;
-            if active.outer_published
-                && retirement::abandoned(&active.ticket, snapshot, stamps, state, false)?
-            {
+        // Groups the firmware dropped unexecuted are published again, with their successors.
+        if self.replay_abandoned(host, defer, false)? {
+            return Ok(true);
+        }
+        // Fallback when a live or exhausted group prevents an ordered replay: a spared front group
+        // the firmware retired without executing during another VM's recovery is retired on its
+        // own even with successors queued, so the pair is not quarantined into a sticky EIO. It
+        // fails with ECANCELED, which spared work never reports as device loss; consecutive
+        // abandoned spared successors are retired the same way.
+        let front_spared = self.active.front().is_some_and(|a| {
+            Arc::ptr_eq(&a.packet, packet) && a.packet.completion.spared()
+        });
+        if !self.quarantined && (self.active.len() == 1 || front_spared) {
+            let mut retired = false;
+            loop {
+                let Some(active) = self.active.front() else {
+                    break;
+                };
+                let spared = active.packet.completion.spared();
+                if retired && !spared {
+                    break;
+                }
+                let mut observer = Observer {
+                    memory: &self.memory,
+                    qids: self.qids(),
+                    host,
+                };
+                core::sync::atomic::fence(core::sync::atomic::Ordering::Acquire);
+                let snapshot = observer.snapshot(active.ticket.ordinal)?;
+                let stamps = observer.stamps()?;
+                let (_, state) = observer.host.epoch()?;
+                if !active.outer_published
+                    || !retirement::abandoned(&active.ticket, snapshot, stamps, state, false)?
+                {
+                    break;
+                }
                 let qos = self.qos_owners();
                 let buffer = self.buffer_id();
+                let left = self.active.len() - 1;
                 let active = self.active.front_mut().ok_or(EIO)?;
                 if !active.flist_completed {
                     host.complete_render(
@@ -1416,19 +1503,30 @@ impl Pair {
                     active.flist_completed = true;
                     active.qos_completed = true;
                 }
-                if !active.failed && !active.packet.completion.spared() {
+                if !active.failed && !spared {
                     active.packet.completion.status().record(EIO);
                     if self.terminal.is_none() {
                         self.terminal = Some((active.packet.completion.status().clone(), error));
                     }
                 }
+                pr_warn!(
+                    "G17: render retired unexecuted after a recovery (ordinal {}, {} groups left)\n",
+                    active.ticket.ordinal,
+                    left
+                );
                 defer(Deferred::Retired(
                     active.packet.completion.clone(),
-                    Err(EIO),
+                    Err(if spared { ECANCELED } else { EIO }),
                 ))?;
                 let ordinal = active.ticket.ordinal;
                 self.memory.retire(ordinal)?;
                 self.active.pop().ok_or(EIO)?;
+                retired = true;
+                if !spared {
+                    break;
+                }
+            }
+            if retired {
                 return Ok(true);
             }
         }
@@ -1456,8 +1554,8 @@ impl Pair {
             self.memory.retire(ordinal)?;
             self.active.pop().ok_or(EIO)?;
         }
-        if let Some(packet) = self.replay.take() {
-            self.replay_since = None;
+        self.replay_since = None;
+        while let Some(packet) = self.replay.pop() {
             defer(Deferred::Drained(packet.completion.clone(), Err(EIO)))?;
         }
         if let Some(previous) = self.previous.as_mut() {
@@ -1620,7 +1718,7 @@ impl Pair {
     /// not left the pair yet.
     pub(crate) fn switch_in_flight(&self) -> Option<u64> {
         let (from, first) = self.switch?;
-        (self.replay.is_some() || self.active.iter().any(|active| active.ticket.ordinal == first))
+        (self.replay.len() != 0 || self.active.iter().any(|active| active.ticket.ordinal == first))
             .then_some(from)
     }
     pub(crate) fn take_retired_binding(&mut self) -> Option<RetiredBinding> {
diff --git a/drivers/gpu/drm/asahi/g17/queue/render/parked.rs b/drivers/gpu/drm/asahi/g17/queue/render/parked.rs
index 5b3f2a00bad6d..41826c03d5201 100644
--- a/drivers/gpu/drm/asahi/g17/queue/render/parked.rs
+++ b/drivers/gpu/drm/asahi/g17/queue/render/parked.rs
@@ -240,7 +240,7 @@ impl ParkedPair {
             terminal: None,
             previous: self.previous.take(),
             displaced: None,
-            replay: None,
+            replay: Tracker::new(),
             replay_since: None,
             context_update: true,
             executed: self.executed,
diff --git a/drivers/gpu/drm/asahi/g17/replay.rs b/drivers/gpu/drm/asahi/g17/replay.rs
new file mode 100644
index 0000000000000..563c3d3f40080
--- /dev/null
+++ b/drivers/gpu/drm/asahi/g17/replay.rs
@@ -0,0 +1,95 @@
+// SPDX-License-Identifier: GPL-2.0-only OR MIT
+
+//! Whether a pair's retained render groups, retired by firmware without executing, are
+//! published again. Only the whole set is replayed, so no replayed group runs after a live
+//! successor, and work a recovery blamed is never run again.
+
+/// One retained group, front first.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) struct Group {
+    /// Published, not failed and not blamed (guilty) by a recovery.
+    pub(crate) innocent: bool,
+    /// Carries the strict abandonment witness.
+    pub(crate) abandoned: bool,
+    /// Already republished the maximum number of times.
+    pub(crate) exhausted: bool,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum Decision {
+    Replay,
+    /// A group may still execute: look again later.
+    Wait,
+    /// Never replay this set; the render timeout settles it.
+    Refuse,
+}
+
+/// Groups are read in order and the first one that decides stops the walk, so a caller can
+/// observe the abandonment lazily. An exhausted front is replayed: the replay fails it alone.
+/// An exhausted successor would break the order.
+pub(crate) fn decide(groups: impl Iterator<Item = Group>) -> Decision {
+    let mut front_exhausted = None;
+    let mut successor_exhausted = false;
+    for group in groups {
+        if !group.innocent {
+            return Decision::Refuse;
+        }
+        if !group.abandoned {
+            return Decision::Wait;
+        }
+        match front_exhausted {
+            None => front_exhausted = Some(group.exhausted),
+            Some(_) => successor_exhausted |= group.exhausted,
+        }
+    }
+    match front_exhausted {
+        Some(true) => Decision::Replay,
+        Some(false) if !successor_exhausted => Decision::Replay,
+        _ => Decision::Refuse,
+    }
+}
+
+#[cfg_attr(not(test), kernel::macros::kunit_tests(asahi_g17_render_replay))]
+mod tests {
+    use super::*;
+    fn group(innocent: bool, abandoned: bool, exhausted: bool) -> Group {
+        Group { innocent, abandoned, exhausted }
+    }
+    const OK: Group = Group { innocent: true, abandoned: true, exhausted: false };
+    #[test]
+    fn every_group_abandoned_and_innocent_is_replayed() {
+        assert_eq!(decide([OK].into_iter()), Decision::Replay);
+        assert_eq!(decide([OK, OK, OK].into_iter()), Decision::Replay);
+    }
+    #[test]
+    fn nothing_retained_is_never_replayed() {
+        assert_eq!(decide([].into_iter()), Decision::Refuse);
+    }
+    #[test]
+    fn guilty_work_is_never_replayed() {
+        assert_eq!(decide([group(false, true, false)].into_iter()), Decision::Refuse);
+        assert_eq!(decide([OK, group(false, true, false)].into_iter()), Decision::Refuse);
+    }
+    #[test]
+    fn a_live_group_waits_and_stops_the_walk() {
+        assert_eq!(decide([OK, group(true, false, false)].into_iter()), Decision::Wait);
+        // Nothing after the live group is looked at, guilty or not.
+        assert_eq!(
+            decide([group(true, false, false), group(false, true, false)].into_iter()),
+            Decision::Wait
+        );
+        let mut seen = 0;
+        let groups = [group(true, false, false), OK, OK].into_iter().inspect(|_| seen += 1);
+        assert_eq!(decide(groups), Decision::Wait);
+        assert_eq!(seen, 1);
+    }
+    #[test]
+    fn an_exhausted_successor_refuses_an_exhausted_front_replays() {
+        assert_eq!(decide([OK, group(true, true, true)].into_iter()), Decision::Refuse);
+        assert_eq!(decide([group(true, true, true), OK].into_iter()), Decision::Replay);
+        assert_eq!(
+            decide([group(true, true, true), group(true, true, true)].into_iter()),
+            Decision::Replay
+        );
+    }
+}
diff --git a/drivers/gpu/drm/asahi/g17/runtime.rs b/drivers/gpu/drm/asahi/g17/runtime.rs
index a9622d5909499..8704b4c5f6e4b 100644
--- a/drivers/gpu/drm/asahi/g17/runtime.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime.rs
@@ -1053,7 +1053,7 @@ struct ComputeHost<'a> {
 }
 impl compute::Host for ComputeHost<'_> {
     fn next_compute_publication(&mut self) -> Result<u64> {
-        self.qids.next_compute_publication()
+        self.qids.next_publication()
     }
     fn epoch(&self) -> Result<(u64, u32)> {
         Ok((
diff --git a/drivers/gpu/drm/asahi/g17/runtime/backend.rs b/drivers/gpu/drm/asahi/g17/runtime/backend.rs
index 0d77ef4a54476..1451b8a75ccc5 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/backend.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/backend.rs
@@ -230,11 +230,30 @@ impl job::Backend for Backend {
         if packet.engine() == Engine::Compute {
             return self.defer_compute_timeout(packet, renewals);
         }
-        if !crate::g17::timeout::transaction_grace(renewals) { return false; }
+        use crate::g17::timeout::{self, Decision};
         let mut state = self.shared.state.lock();
-        (*state)
-            .as_deref_mut()
-            .is_some_and(|firmware| firmware.queues.defer_render_timeout(packet, ETIMEDOUT))
+        let Some(firmware) = (*state).as_deref_mut() else {
+            return false;
+        };
+        // Borrowed-graph and queued-retirement deferrals share the renewal bound.
+        if timeout::transaction_grace(renewals)
+            && firmware.queues.defer_render_timeout(packet, ETIMEDOUT)
+        {
+            return true;
+        }
+        let decision = firmware.queues.render_timeout_decision(packet, renewals);
+        // As compute does: work still blocked behind another VM's older unfinished work after the
+        // whole grace is innocent, so its failure must stay non-sticky. The pair's quarantine
+        // extends the class to the pair's other groups.
+        if decision == Decision::Spare {
+            packet.completion.classify(crate::g17::recovery::Class::Spared);
+        }
+        if decision != Decision::Own && (renewals == 0 || decision == Decision::Spare) {
+            dev_warn!(self.shared.dev.as_ref(),
+                "Render watchdog owner={} renewals={} decision={:?} behind earlier work of another VM\n",
+                self.owner, renewals, decision);
+        }
+        decision == Decision::Defer
     }
 
     fn try_early(&self, packet: &Arc<Packet>) -> bool {
diff --git a/drivers/gpu/drm/asahi/g17/runtime/compute_control.rs b/drivers/gpu/drm/asahi/g17/runtime/compute_control.rs
index 80f88d09191b4..21dd1515cd59b 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/compute_control.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/compute_control.rs
@@ -27,10 +27,11 @@ impl Backend {
             let Some(own) = queue.timeout_head() else {
                 return Ok(Decision::Own);
             };
-            let others = firmware.queues.compute.iter().flatten()
+            let computes = firmware.queues.compute.iter().flatten()
                 .filter_map(|entry| entry.queue.as_deref())
                 .filter(|other| other.owner() != Some(self.owner))
                 .filter_map(|other| other.timeout_head());
+            let others = Iterator::chain(computes, firmware.queues.render_timeout_heads());
             let decision = timeout::classify(own, renewals, others);
             if decision == Decision::Spare {
                 firmware.queues.compute(self.owner).ok_or(EIO)?
diff --git a/drivers/gpu/drm/asahi/g17/runtime/render.rs b/drivers/gpu/drm/asahi/g17/runtime/render.rs
index 88340442bbcae..876585fb244e4 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/render.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/render.rs
@@ -546,6 +546,9 @@ impl crate::g17::queue::render::Host for Host<'_> {
     fn publish_qid(&mut self, id: kick::Id) -> Result {
         self.firmware.queues.qids.publish(id)
     }
+    fn next_publication(&mut self) -> Result<u64> {
+        self.firmware.queues.qids.next_publication()
+    }
     fn publish_outer_pair(
         &mut self,
         priority: u8,
@@ -953,6 +956,52 @@ impl Registry {
         entry.pending_timeout = true;
         true
     }
+
+    /// The oldest published group of every render pair, for cross-VM timeout attribution
+    /// (crate::g17::timeout).
+    pub(in crate::g17) fn render_timeout_heads(
+        &self,
+    ) -> impl Iterator<Item = crate::g17::timeout::Head> + '_ {
+        self.render
+            .entries
+            .iter()
+            .flatten()
+            .filter_map(|entry| entry.pair.as_ref())
+            .filter_map(|pair| pair.timeout_head())
+    }
+
+    /// A render watchdog blocked behind another VM's older unfinished work gets the same
+    /// bounded grace as compute; the oldest unfinished work is never excused. When the GPU
+    /// stalls on one client, the others' renders then outlive the culprit's teardown instead
+    /// of all timing out together.
+    pub(in crate::g17) fn render_timeout_decision(
+        &self,
+        packet: &Arc<Packet>,
+        renewals: u8,
+    ) -> crate::g17::timeout::Decision {
+        let own = self
+            .render
+            .entries
+            .iter()
+            .flatten()
+            .filter_map(|entry| entry.pair.as_ref())
+            .find(|pair| pair.contains_packet(packet))
+            .and_then(|pair| pair.timeout_head());
+        let computes = self
+            .compute
+            .iter()
+            .flatten()
+            .filter_map(|entry| entry.queue.as_deref())
+            .filter_map(|queue| queue.timeout_head());
+        match own {
+            Some(own) => crate::g17::timeout::classify(
+                own,
+                renewals,
+                Iterator::chain(self.render_timeout_heads(), computes),
+            ),
+            None => crate::g17::timeout::Decision::Own,
+        }
+    }
 }
 
 impl RenderState {
diff --git a/drivers/gpu/drm/asahi/g17/runtime/render/replay.rs b/drivers/gpu/drm/asahi/g17/runtime/render/replay.rs
index bf5ba6caa7a79..499e3858fcca1 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/render/replay.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/render/replay.rs
@@ -2,8 +2,8 @@
 
 //! Republication of render groups that firmware retired without executing.
 //! The pair proves the abandonment (retired, unlinked, stamps covering it, no
-//! recovery, fragment end never written) and parks the packet; this pass
-//! prepares and publishes the same command again under the device mutex.
+//! recovery, fragment end never written) and parks the packets in order; this
+//! pass prepares and publishes the same commands again under the device mutex.
 
 use super::super::DeferredBatch;
 use crate::g17::{object::Allocator, queue::render::RenderAddresses, Firmware};
@@ -27,81 +27,91 @@ impl Firmware {
             return Ok(());
         }
         for slot in 0..self.queues.render.entries.len() {
-            let Some((packet, owner, qids, closed, busy, stalled)) =
-                self.queues.render.entries[slot].as_ref().and_then(|entry| {
-                    let pair = entry.pair.as_ref()?;
-                    let packet = pair.replay_front()?;
-                    Some((
-                        packet,
-                        entry.reservation.owner,
-                        pair.qids(),
-                        entry.closed,
-                        entry.preparing.is_some()
-                            || entry.growing
-                            || entry.pending_timeout
-                            || entry.pending_error.is_some(),
-                        pair.replay_stalled(crate::g17::now_ns()),
-                    ))
-                })
-            else {
-                continue;
-            };
-            if stalled { return Err(ETIMEDOUT); }
-            if busy {
-                continue;
+            // Publish the parked groups front first; a pause keeps the rest parked.
+            while self.replay_render_front(slot, deferred)? {}
+        }
+        Ok(())
+    }
+
+    /// Publishes the oldest parked group of `slot`; returns whether it was published.
+    fn replay_render_front(&mut self, slot: usize, deferred: &mut DeferredBatch) -> Result<bool> {
+        let Some((packet, owner, qids, closed, busy, stalled)) =
+            self.queues.render.entries[slot].as_ref().and_then(|entry| {
+                let pair = entry.pair.as_ref()?;
+                let packet = pair.replay_front()?;
+                Some((
+                    packet,
+                    entry.reservation.owner,
+                    pair.qids(),
+                    entry.closed,
+                    entry.preparing.is_some()
+                        || entry.growing
+                        || entry.pending_timeout
+                        || entry.pending_error.is_some(),
+                    pair.replay_stalled(crate::g17::now_ns()),
+                ))
+            })
+        else {
+            return Ok(false);
+        };
+        if stalled {
+            return Err(ETIMEDOUT);
+        }
+        if busy {
+            return Ok(false);
+        }
+        if closed || packet.cancel_requested() || !packet.context.is_current() {
+            self.with_render(slot, |pair, _| {
+                pair.fail_replay(ECANCELED, &mut |event| deferred.push(event))
+            })?;
+            return Ok(false);
+        }
+        if packet.completion.status().get() != 0 {
+            self.with_render(slot, |pair, _| {
+                pair.fail_replay(EIO, &mut |event| deferred.push(event))
+            })?;
+            return Ok(false);
+        }
+        let addresses = RenderAddresses {
+            stamps: [
+                self.init.stamp_addresses(qids[0])?,
+                self.init.stamp_addresses(qids[1])?,
+            ],
+        };
+        let dev = self.primary.state.shared.drm()?;
+        let uat = self.uat.clone();
+        let prepared = self.with_render(slot, |pair, _| {
+            if !pair.room() {
+                return Err(EBUSY);
             }
-            if closed || packet.cancel_requested() || !packet.context.is_current() {
+            let alloc = Allocator {
+                dev: &dev,
+                uat: &uat,
+            };
+            pair.prepare(&alloc, &packet, addresses)
+        });
+        match prepared {
+            Ok(()) => {}
+            Err(error) if error == EBUSY || error == EAGAIN => return Ok(false),
+            Err(error) => {
                 self.with_render(slot, |pair, _| {
-                    pair.fail_replay(ECANCELED, &mut |event| deferred.push(event))
+                    pair.fail_replay(error, &mut |event| deferred.push(event))
                 })?;
-                continue;
+                return Ok(false);
+            }
+        }
+        match self.publish_render(owner, slot as u8, &packet, deferred) {
+            Ok(()) => {
+                self.with_render(slot, |pair, _| pair.finish_replay(&packet))?;
+                Ok(true)
             }
-            if packet.completion.status().get() != 0 {
+            Err(error) if error == EBUSY || error == EAGAIN => Ok(false),
+            Err(error) => {
                 self.with_render(slot, |pair, _| {
-                    pair.fail_replay(EIO, &mut |event| deferred.push(event))
+                    pair.fail_replay(error, &mut |event| deferred.push(event))
                 })?;
-                continue;
-            }
-            let addresses = RenderAddresses {
-                stamps: [
-                    self.init.stamp_addresses(qids[0])?,
-                    self.init.stamp_addresses(qids[1])?,
-                ],
-            };
-            let dev = self.primary.state.shared.drm()?;
-            let uat = self.uat.clone();
-            let prepared = self.with_render(slot, |pair, _| {
-                if !pair.room() {
-                    return Err(EBUSY);
-                }
-                let alloc = Allocator {
-                    dev: &dev,
-                    uat: &uat,
-                };
-                pair.prepare(&alloc, &packet, addresses)
-            });
-            match prepared {
-                Ok(()) => {}
-                Err(error) if error == EBUSY || error == EAGAIN => continue,
-                Err(error) => {
-                    self.with_render(slot, |pair, _| {
-                        pair.fail_replay(error, &mut |event| deferred.push(event))
-                    })?;
-                    continue;
-                }
-            }
-            match self.publish_render(owner, slot as u8, &packet, deferred) {
-                Ok(()) => {
-                    self.with_render(slot, |pair, _| pair.finish_replay(&packet))?;
-                }
-                Err(error) if error == EBUSY || error == EAGAIN => continue,
-                Err(error) => {
-                    self.with_render(slot, |pair, _| {
-                        pair.fail_replay(error, &mut |event| deferred.push(event))
-                    })?;
-                }
+                Ok(false)
             }
         }
-        Ok(())
     }
 }
diff --git a/drivers/gpu/drm/asahi/g17/runtime/render/settle.rs b/drivers/gpu/drm/asahi/g17/runtime/render/settle.rs
index 5db82a08bc8d1..d93e701b8b5cb 100644
--- a/drivers/gpu/drm/asahi/g17/runtime/render/settle.rs
+++ b/drivers/gpu/drm/asahi/g17/runtime/render/settle.rs
@@ -125,6 +125,12 @@ impl crate::g17::Firmware {
             .find(|pair| pair.qids().contains(&qid))
             .and_then(|pair| pair.oldest_packet());
         if let Some(packet) = packet {
+            // Unclassified work counts as innocent, and a blamed group must never be replayed.
+            // Work an earlier recovery spared is collateral here (classify() blames before its
+            // collateral check) and stays spared.
+            if !packet.completion.spared() {
+                packet.completion.classify(crate::g17::recovery::Class::Guilty);
+            }
             self.fail_render_packet(&packet, error, deferred)?;
         }
         Ok(())
diff --git a/drivers/gpu/drm/asahi/sysfs.c b/drivers/gpu/drm/asahi/sysfs.c
index b6ab692ce3f54..7d7d785a83133 100644
--- a/drivers/gpu/drm/asahi/sysfs.c
+++ b/drivers/gpu/drm/asahi/sysfs.c
@@ -170,6 +170,46 @@ int asahi_sysfs_register(struct device *dev, const void *snapshot,
 }
 EXPORT_SYMBOL_GPL(asahi_sysfs_register);
 
+/*
+ * G17: GPU idle power-off delay (ms, 2-1000). On T6040/25G83 the firmware powers the GPU cores
+ * off after 2 ms idle and reads its own delay only at boot; above 2 the driver holds the cores
+ * powered until the GPU has been idle this long (power::IDLE_HOLD_MS). Each wake-up there costs
+ * about 1 ms, so userspace may lengthen it on AC power.
+ */
+static u32 *asahi_idle_word;
+
+static ssize_t idle_off_delay_ms_show(struct device *dev, struct device_attribute *attr, char *buf)
+{
+	return sysfs_emit(buf, "%u\n", READ_ONCE(*asahi_idle_word));
+}
+
+static ssize_t idle_off_delay_ms_store(struct device *dev, struct device_attribute *attr,
+				       const char *buf, size_t count)
+{
+	u32 ms;
+
+	if (kstrtou32(buf, 0, &ms) || ms < 2 || ms > 1000)
+		return -EINVAL;
+	WRITE_ONCE(*asahi_idle_word, ms);
+	return count;
+}
+
+static DEVICE_ATTR_RW(idle_off_delay_ms);
+
+int asahi_sysfs_register_idle(struct device *dev, u32 *word)
+{
+	int ret;
+
+	if (!dev || !word)
+		return -ENODEV;
+	asahi_idle_word = word;
+	ret = device_create_file(dev, &dev_attr_idle_off_delay_ms);
+	if (ret)
+		asahi_idle_word = NULL;
+	return ret;
+}
+EXPORT_SYMBOL_GPL(asahi_sysfs_register_idle);
+
 void asahi_sysfs_unregister(struct device *dev, void *handle)
 {
 	struct asahi_stats_attribute *stats = handle;
@@ -180,6 +220,10 @@ void asahi_sysfs_unregister(struct device *dev, void *handle)
 	/* Removal drains active show callbacks before either allocation dies. */
 	device_remove_file(dev, &stats->memory_attr);
 	device_remove_file(dev, &stats->attr);
+	if (asahi_idle_word) {
+		device_remove_file(dev, &dev_attr_idle_off_delay_ms);
+		asahi_idle_word = NULL;
+	}
 	kfree(stats);
 }
 EXPORT_SYMBOL_GPL(asahi_sysfs_unregister);
diff --git a/drivers/gpu/drm/asahi/sysfs.h b/drivers/gpu/drm/asahi/sysfs.h
index 0ef6b5c3593bc..f1873f9be6caa 100644
--- a/drivers/gpu/drm/asahi/sysfs.h
+++ b/drivers/gpu/drm/asahi/sysfs.h
@@ -7,6 +7,7 @@ struct device;
 
 int asahi_sysfs_register(struct device *dev, const void *snapshot,
 			 int export_enabled, const void *memory, void **handle);
+int asahi_sysfs_register_idle(struct device *dev, u32 *word);
 void asahi_sysfs_unregister(struct device *dev, void *handle);
 
 #endif /* _ASAHI_SYSFS_H */
diff --git a/drivers/gpu/drm/asahi/sysfs_exports.rs b/drivers/gpu/drm/asahi/sysfs_exports.rs
index d8c03b1d5d6ff..cee5d52a4e261 100644
--- a/drivers/gpu/drm/asahi/sysfs_exports.rs
+++ b/drivers/gpu/drm/asahi/sysfs_exports.rs
@@ -17,6 +17,21 @@ extern "C" {
     fn asahi_sysfs_unregister(dev: *mut c_void, handle: *mut c_void);
 }
 
+extern "C" {
+    fn asahi_sysfs_register_idle(dev: *mut c_void, word: *mut u32) -> c_int;
+}
+
+/// Export the idle power-off delay as the read-write `idle_off_delay_ms` (G17).
+pub(crate) fn register_idle(
+    dev: *mut kernel::bindings::device,
+    word: *mut u32,
+) -> kernel::error::Result {
+    // SAFETY: `word` is a static atomic that outlives the attribute; the C side reads and writes
+    // it with READ_ONCE/WRITE_ONCE.
+    let ret = unsafe { asahi_sysfs_register_idle(dev.cast(), word) };
+    if ret < 0 { Err(kernel::error::Error::from_errno(ret)) } else { Ok(()) }
+}
+
 /// Register only after initialization succeeds, keeping the snapshot Arc alive.
 ///
 /// Also registers `gpu_memory`, which shows the static GPU memory ledger.