diff --git a/crates/perry-runtime/src/agent.rs b/crates/perry-runtime/src/agent.rs
new file mode 100644
index 0000000000..d61e147ca3
--- /dev/null
+++ b/crates/perry-runtime/src/agent.rs
@@ -0,0 +1,163 @@
+//! Agent (JS heap owner) identity for cross-thread queue dispatch. Issue #6185
+//! Tier 2.
+//!
+//! ## The problem this solves
+//!
+//! Perry's cross-thread plumbing keeps three process-global queues —
+//! `PENDING_THREAD_RESULTS` (`thread.rs`), `TIMER_QUEUE` / `CALLBACK_TIMERS` /
+//! `INTERVAL_TIMERS` (`timer.rs`) — whose entries carry raw pointers
+//! (`*mut Promise`, closure handles, argument `f64`s that may be NaN-boxed
+//! pointers) into *some* thread's arena. Their drains were unconditional: any
+//! thread that ran the await loop / microtask pump drained *everything*. A
+//! `perry/thread` worker doing async work could therefore steal a main-thread
+//! completion, deserialize it into the worker's arena, resolve a main-heap
+//! promise with a worker-heap pointer, and fire main-heap timer closures on the
+//! worker. When the worker exits, its arena is unmapped and the main thread
+//! reads freed memory. The `unsafe impl Send` on those queue entries carried a
+//! "only accessed from the pump thread" comment that nothing enforced.
+//!
+//! ## Why not tag with `ThreadId`
+//!
+//! The obvious fix — tag each entry with the `ThreadId` that enqueued it and
+//! have drains skip entries they don't own — breaks Android. There, the
+//! compiled TypeScript (and therefore every `setTimeout` / `setInterval` call)
+//! runs on the `perry-native` thread, but the timer pump fires from the UI
+//! thread via `nativePumpTick` (`perry-ui-android/src/app.rs`). A blanket
+//! owner-skip would leave every Android timer un-fired forever. That's the
+//! blocker that deferred the narrow slice of this fix earlier.
+//!
+//! ## The model: agents, and pumps that act on their behalf
+//!
+//! The unit that matters is not the OS thread but the **heap** — the arena the
+//! queued pointers live in. We call it an *agent*.
+//!
+//! - The program starts with one agent, [`PRIMARY_AGENT`]. Every thread that
+//!   runs user JS on the primary heap (the process main thread; Android's
+//!   `perry-native` thread) belongs to it.
+//! - Each `perry/thread` worker gets its **own** agent id at thread entry (see
+//!   [`enter_worker_agent`]), because it gets its own arena and GC.
+//! - A thread that has *not* been given an agent of its own is, by definition,
+//!   not a JS heap owner: it is a **pump** running on behalf of the primary
+//!   agent. Android's UI thread is exactly this. So [`current_agent`] falls back
+//!   to [`PRIMARY_AGENT`] rather than inventing an id.
+//!
+//! That fallback is what makes the Android path keep working unchanged: the UI
+//! thread resolves to `PRIMARY_AGENT`, which is precisely the agent whose timers
+//! `perry-native` scheduled. Meanwhile a worker resolves to its own id, so its
+//! drains can only ever touch entries it enqueued itself.
+//!
+//! Queue entries are tagged with [`current_agent`] at *enqueue* time and drains
+//! filter on [`current_agent`] at *drain* time. Foreign entries are left in
+//! place for their owner (never silently dropped mid-flight); entries belonging
+//! to an agent that has exited are purged by [`retire_agent`], since that
+//! agent's arena is gone and nothing can ever settle them.
+
+use std::cell::Cell;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+/// Identity of a JS heap (arena + GC) that queued cross-thread work.
+pub type AgentId = u64;
+
+/// The agent that owns the process's initial JS heap: the main thread, and on
+/// Android the `perry-native` thread. Threads with no agent of their own (pump
+/// threads such as Android's UI thread) act on this agent's behalf.
+pub const PRIMARY_AGENT: AgentId = 0;
+
+/// Source of worker agent ids. Starts at 1 — 0 is reserved for
+/// [`PRIMARY_AGENT`], which is never handed out here.
+static NEXT_AGENT: AtomicU64 = AtomicU64::new(1);
+
+thread_local! {
+    /// The agent this thread *owns*. `None` means "no heap of my own" — the
+    /// main/`perry-native` JS thread (which owns the primary heap implicitly)
+    /// and any pure pump thread both leave this unset, and so resolve to
+    /// [`PRIMARY_AGENT`] via [`current_agent`].
+    static CURRENT_AGENT: Cell<Option<AgentId>> = const { Cell::new(None) };
+}
+
+/// Claim a fresh agent id for the calling thread. Called once at the top of
+/// every `perry/thread` worker (`spawn` / `parallelMap` / `parallelFilter`),
+/// before the worker can allocate or enqueue anything, so every pointer it
+/// puts into a global queue is tagged as its own.
+///
+/// Returns the new id so the caller can hand it to [`retire_agent`] at exit.
+pub fn enter_worker_agent() -> AgentId {
+    let id = NEXT_AGENT.fetch_add(1, Ordering::Relaxed);
+    CURRENT_AGENT.with(|slot| slot.set(Some(id)));
+    id
+}
+
+/// The agent whose queued work the calling thread may touch: its own if it is a
+/// worker, otherwise [`PRIMARY_AGENT`] (it is a JS thread on the primary heap,
+/// or a pump acting for it).
+pub fn current_agent() -> AgentId {
+    CURRENT_AGENT
+        .with(|slot| slot.get())
+        .unwrap_or(PRIMARY_AGENT)
+}
+
+/// True when `owner` is work the calling thread is allowed to drain and run.
+#[inline]
+pub fn owns(owner: AgentId) -> bool {
+    owner == current_agent()
+}
+
+/// Retire a worker agent at thread exit: its arena is about to be unmapped, so
+/// any queue entry still tagged with it points at memory that is about to go
+/// away. Purge those entries rather than leaving them for a drain that can
+/// never legally run.
+///
+/// Purging is safe precisely because nothing else can own them: the entries name
+/// pointers into *this* agent's arena, and no other thread may dereference them.
+pub fn retire_agent(id: AgentId) {
+    debug_assert_ne!(
+        id, PRIMARY_AGENT,
+        "the primary agent outlives the process; it is never retired"
+    );
+    crate::timer::purge_agent_timers(id);
+    crate::thread::purge_agent_thread_results(id);
+    // Deliberately do NOT clear `CURRENT_AGENT`. Clearing it would make
+    // `current_agent()` fall back to `PRIMARY_AGENT` for the rest of this
+    // thread's life — i.e. a worker that has just torn down its heap would
+    // start reporting itself as an owner of the *main* thread's queued work,
+    // reintroducing exactly the cross-heap drain this fix removes. A retired
+    // worker keeps its (now dead) id and therefore owns nothing at all.
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A plain thread that never claims a worker agent — the main thread, or a
+    /// pump like Android's UI thread — resolves to the primary agent, so it
+    /// keeps draining exactly the work the primary JS thread scheduled. This is
+    /// the property that keeps `nativePumpTick` delivering Android timers.
+    #[test]
+    fn unclaimed_threads_pump_for_the_primary_agent() {
+        assert_eq!(current_agent(), PRIMARY_AGENT);
+        assert!(owns(PRIMARY_AGENT));
+
+        let pump_thread_agent = std::thread::spawn(current_agent).join().unwrap();
+        assert_eq!(pump_thread_agent, PRIMARY_AGENT);
+    }
+
+    /// A worker claims its own agent, so it neither drains the primary agent's
+    /// work nor collides with a sibling worker.
+    #[test]
+    fn workers_get_distinct_agents_and_disown_primary_work() {
+        let a = std::thread::spawn(|| {
+            let id = enter_worker_agent();
+            assert_eq!(current_agent(), id);
+            assert!(!owns(PRIMARY_AGENT), "a worker must not own primary work");
+            id
+        })
+        .join()
+        .unwrap();
+
+        let b = std::thread::spawn(|| enter_worker_agent()).join().unwrap();
+
+        assert_ne!(a, PRIMARY_AGENT);
+        assert_ne!(b, PRIMARY_AGENT);
+        assert_ne!(a, b, "sibling workers must not share an agent");
+    }
+}
diff --git a/crates/perry-runtime/src/agent_dispatch_tests.rs b/crates/perry-runtime/src/agent_dispatch_tests.rs
new file mode 100644
index 0000000000..dd15c878e2
--- /dev/null
+++ b/crates/perry-runtime/src/agent_dispatch_tests.rs
@@ -0,0 +1,202 @@
+//! #6185 Tier 2 — owner-tagged cross-thread dispatch.
+//!
+//! These exercise the three process-global queues the 2026-07-09 GC audit
+//! flagged (`PENDING_THREAD_RESULTS`, `TIMER_QUEUE` / `CALLBACK_TIMERS` /
+//! `INTERVAL_TIMERS`) at the Rust level: an entry enqueued by one agent must be
+//! invisible to every other agent's drain, and must survive — not be silently
+//! eaten — when a foreign agent pumps.
+//!
+//! The end-to-end JS behavior (a worker's `await` no longer stealing the main
+//! thread's `spawn` completion) rides on top of this and is covered by the
+//! compiled integration test in
+//! `crates/perry/tests/issue_6185_thread_owner_tagged_dispatch.rs`.
+
+use crate::agent::{current_agent, enter_worker_agent, owns, retire_agent, PRIMARY_AGENT};
+
+/// The timer queues are process-global, and `cargo test` runs these in parallel
+/// threads of one process — so two tests that both schedule on the PRIMARY agent
+/// would see each other's timers in `active_timeout_resource_count()`. Serialize
+/// the ones that assert on primary-agent counts.
+static TIMER_QUEUE_TESTS: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
+/// The Android invariant, stated as a test: `nativePumpTick` runs on the UI
+/// thread, which never claims a worker agent, so it must resolve to the same
+/// agent as the `perry-native` thread that scheduled the timers. If this ever
+/// regresses to a raw-`ThreadId` comparison, every Android timer stops firing —
+/// the exact blocker that deferred this fix the first time round.
+#[test]
+fn a_pump_thread_shares_the_primary_agent_with_the_js_thread() {
+    // The "perry-native" thread: runs JS, never claims a worker agent.
+    let js_thread = std::thread::spawn(current_agent).join().unwrap();
+    // The UI thread: pure pump, also never claims a worker agent.
+    let ui_pump_thread = std::thread::spawn(current_agent).join().unwrap();
+
+    assert_eq!(js_thread, PRIMARY_AGENT);
+    assert_eq!(
+        ui_pump_thread, js_thread,
+        "the UI pump must own the JS thread's work, or Android timers never fire"
+    );
+    // And it can in fact claim that work.
+    assert!(std::thread::spawn(|| owns(PRIMARY_AGENT)).join().unwrap());
+}
+
+/// A worker owns its own agent and disowns the primary agent's work — this is
+/// what stops a worker's await loop from draining the main thread's queues.
+#[test]
+fn a_worker_disowns_the_primary_agents_work() {
+    let (worker_agent, worker_owns_primary) = std::thread::spawn(|| {
+        let id = enter_worker_agent();
+        let owns_primary = owns(PRIMARY_AGENT);
+        retire_agent(id);
+        (id, owns_primary)
+    })
+    .join()
+    .unwrap();
+
+    assert_ne!(worker_agent, PRIMARY_AGENT);
+    assert!(
+        !worker_owns_primary,
+        "a worker draining primary-agent work is the #6185 use-after-free"
+    );
+}
+
+/// A timer scheduled by the primary agent must not be *fired* by a worker — and
+/// must still be there afterwards for its real owner. The pre-fix drain would
+/// have run the callback on the worker (cross-heap call) or, with a naive
+/// filter-and-drop, silently eaten it.
+#[test]
+fn a_worker_neither_fires_nor_eats_a_primary_agent_timer() {
+    let _serial = TIMER_QUEUE_TESTS.lock().unwrap_or_else(|e| e.into_inner());
+    // Schedule on the primary agent (this test thread).
+    let before = crate::timer::active_timeout_resource_count();
+    let id = crate::timer::js_set_timeout_callback(0, 50_000.0);
+    assert_eq!(
+        crate::timer::active_timeout_resource_count(),
+        before + 1,
+        "the owning agent sees its own timer as an active handle"
+    );
+
+    // A worker pumps the timer queues. It must fire nothing of ours...
+    let fired_on_worker = std::thread::spawn(|| {
+        let agent = enter_worker_agent();
+        let fired = crate::timer::js_callback_timer_tick() + crate::timer::js_interval_timer_tick();
+        // ...and must not count our timer as work keeping ITS loop alive.
+        let visible = crate::timer::active_timeout_resource_count();
+        let has_pending = crate::timer::js_callback_timer_has_pending();
+        retire_agent(agent);
+        (fired, visible, has_pending)
+    })
+    .join()
+    .unwrap();
+
+    assert_eq!(fired_on_worker.0, 0, "a worker must not fire our timers");
+    assert_eq!(
+        fired_on_worker.1, 0,
+        "a worker must not see our timer as its active handle"
+    );
+    assert_eq!(
+        fired_on_worker.2, 0,
+        "a foreign timer must not keep a worker's event loop alive"
+    );
+
+    // The timer is still ours, untouched, after the foreign pump.
+    assert_eq!(
+        crate::timer::active_timeout_resource_count(),
+        before + 1,
+        "a foreign pump must not consume the owner's timer"
+    );
+
+    crate::timer::clearTimeout(id);
+}
+
+/// The GC root scanner must walk only the collecting agent's own timers.
+///
+/// This is the subtlest half of the fix. `scan_timer_roots_mut` runs on every GC
+/// cycle and, on an *evacuating* cycle, REWRITES the slots it visits to
+/// forwarding addresses in the collecting thread's arena. Walking a foreign
+/// agent's timer there would mark through a heap this collector does not own
+/// (racing that agent's collector on the same `GcHeader` bits) and rewrite its
+/// promise/closure slots to point into the wrong arena.
+///
+/// So a worker collecting while the primary agent holds a timer must visit zero
+/// of the primary agent's slots — while the owner still roots its own.
+#[test]
+fn a_workers_gc_scan_never_visits_another_agents_timer_slots() {
+    let _serial = TIMER_QUEUE_TESTS.lock().unwrap_or_else(|e| e.into_inner());
+
+    // A promise timer on the primary agent. Its `promise` (a raw pointer into
+    // THIS arena) and `value` slots are scanned unconditionally, which is what
+    // makes it a clean probe for "did the collector touch a foreign slot?".
+    let baseline = {
+        let mut n = 0usize;
+        crate::timer::scan_timer_roots(&mut |_v: f64| n += 1);
+        n
+    };
+    let _promise = crate::timer::js_set_timeout(50_000.0);
+    let visited_by_owner = {
+        let mut n = 0usize;
+        crate::timer::scan_timer_roots(&mut |_v: f64| n += 1);
+        n
+    };
+    assert!(
+        visited_by_owner > baseline,
+        "the owning agent must root its own timer's slots, or the promise is swept"
+    );
+
+    // The same scan on a worker must not see any of it.
+    let visited_on_worker = std::thread::spawn(|| {
+        let agent = enter_worker_agent();
+        let mut visited = 0usize;
+        crate::timer::scan_timer_roots(&mut |_v: f64| visited += 1);
+        retire_agent(agent);
+        visited
+    })
+    .join()
+    .unwrap();
+
+    assert_eq!(
+        visited_on_worker, 0,
+        "a worker's GC must not mark (or evacuate-rewrite) the primary agent's timer slots"
+    );
+
+    crate::timer::test_clear_all_timer_scanner_roots();
+}
+
+/// Retiring a worker purges the timers it scheduled: its arena is unmapped, so
+/// nothing can ever legally run those callbacks. Pre-fix they stayed on the
+/// global queue and fired on whatever thread pumped next — reading freed memory.
+#[test]
+fn retiring_a_worker_purges_the_timers_it_scheduled() {
+    let _serial = TIMER_QUEUE_TESTS.lock().unwrap_or_else(|e| e.into_inner());
+    let before = crate::timer::active_timeout_resource_count();
+
+    std::thread::spawn(|| {
+        let agent = enter_worker_agent();
+        // A worker scheduling a timer: the callback closure would live in this
+        // worker's arena.
+        crate::timer::js_set_timeout_callback(0, 60_000.0);
+        crate::timer::setInterval(0, 60_000.0);
+        assert!(
+            crate::timer::active_timeout_resource_count() >= 2,
+            "the worker sees its own timers while it is alive"
+        );
+        retire_agent(agent);
+        assert_eq!(
+            crate::timer::active_timeout_resource_count(),
+            0,
+            "retiring the agent drops the timers it owned"
+        );
+        assert!(
+            !owns(PRIMARY_AGENT),
+            "a retired worker must not fall back to owning the primary agent's work"
+        );
+    })
+    .join()
+    .unwrap();
+
+    assert_eq!(
+        crate::timer::active_timeout_resource_count(),
+        before,
+        "a dead worker's timers must not linger on the primary agent's queue"
+    );
+}
diff --git a/crates/perry-runtime/src/atomics.rs b/crates/perry-runtime/src/atomics.rs
index ceca7908e5..2b35827d6c 100644
--- a/crates/perry-runtime/src/atomics.rs
+++ b/crates/perry-runtime/src/atomics.rs
@@ -775,13 +775,17 @@ pub extern "C" fn js_atomics_wait_async(
     }
     crate::thread::thread_job_begin();
     let promise_usize = promise as usize;
+    // #6185: the promise belongs to the agent calling `waitAsync`. The futex
+    // waiter below owns no heap (it never runs JS), so it cannot infer the
+    // right agent from itself — capture it here, on the calling thread.
+    let owner_agent = crate::agent::current_agent();
     std::thread::spawn(move || {
         let outcome = crate::atomics_futex::block(handle, deadline);
         let result = match outcome {
             crate::atomics_futex::WaitOutcome::Ok => "ok",
             _ => "timed-out",
         };
-        crate::thread::queue_promise_string_result(promise_usize, result);
+        crate::thread::queue_promise_string_result(owner_agent, promise_usize, result);
     });
     wait_async_result(true, crate::value::js_nanbox_pointer(promise as i64))
 }
diff --git a/crates/perry-runtime/src/lib.rs b/crates/perry-runtime/src/lib.rs
index 8004abdfbe..edfde97e39 100644
--- a/crates/perry-runtime/src/lib.rs
+++ b/crates/perry-runtime/src/lib.rs
@@ -31,6 +31,9 @@ static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
 static GLOBAL: std::alloc::System = std::alloc::System;
 
 pub mod abi_trampoline;
+pub mod agent;
+#[cfg(test)]
+mod agent_dispatch_tests;
 pub mod app_group;
 pub mod arena;
 pub mod array;
diff --git a/crates/perry-runtime/src/thread.rs b/crates/perry-runtime/src/thread.rs
index b617645449..adc5a79171 100644
--- a/crates/perry-runtime/src/thread.rs
+++ b/crates/perry-runtime/src/thread.rs
@@ -912,6 +912,10 @@ unsafe fn parallel_map_impl(array_val: f64, closure_val: f64) -> i64 {
             let captures_ref = captures_arc.clone();
 
             let handle = scope.spawn(move || {
+                // #6185: own agent id before any allocation or enqueue, so this
+                // worker's drains can't touch the spawner's queued work (and
+                // anything it queues is tagged as its own).
+                let worker_agent = crate::agent::enter_worker_agent();
                 // Each thread has its own arena (via thread_local!).
                 // Register this thread's root scanners BEFORE any allocation
                 // can cross a GC trigger — a fresh worker otherwise collects
@@ -953,6 +957,11 @@ unsafe fn parallel_map_impl(array_val: f64, closure_val: f64) -> i64 {
                     results.push(serialize_nanbox_for_thread(result.to_bits()));
                 }
 
+                // #6185: results are already serialized into agent-independent
+                // form; this arena is about to go away with the scope, so purge
+                // anything this worker left in a global queue.
+                drop(gc_scope);
+                crate::agent::retire_agent(worker_agent);
                 (idx, results)
             });
             handles.push(handle);
@@ -1139,9 +1148,11 @@ unsafe fn parallel_filter_impl(array_val: f64, closure_val: f64) -> i64 {
             let captures_ref = captures_arc.clone();
 
             let handle = scope.spawn(move || {
-                // See parallel_map's worker: scanner registration must precede
+                // See parallel_map's worker: own agent (#6185) before anything
+                // can allocate or enqueue, scanner registration must precede
                 // any allocation, and the rebuilt closure must be rooted
                 // across the per-element deserialization allocations.
+                let worker_agent = crate::agent::enter_worker_agent();
                 crate::gc::ensure_gc_initialized();
                 let mut kept = Vec::new();
 
@@ -1178,6 +1189,10 @@ unsafe fn parallel_filter_impl(array_val: f64, closure_val: f64) -> i64 {
                     }
                 }
 
+                // #6185: see parallel_map's worker — purge this agent's queue
+                // entries before its arena goes away.
+                drop(gc_scope);
+                crate::agent::retire_agent(worker_agent);
                 (idx, kept)
             });
             handles.push(handle);
@@ -1317,10 +1332,18 @@ unsafe fn spawn_impl(closure_val: f64) -> *mut crate::promise::Promise {
     (*promise_header).gc_flags |= gc::GC_FLAG_PINNED;
 
     let promise_usize = promise as usize;
+    // #6185: the promise lives in the SPAWNING agent's heap, so that is the
+    // agent allowed to settle it. Captured here, on the spawning thread —
+    // reading it inside the worker would yield the worker's own agent.
+    let owner_agent = crate::agent::current_agent();
 
     // ── 3. Spawn background thread ───────────────────────────────────
     ACTIVE_THREAD_JOBS.fetch_add(1, Ordering::SeqCst);
     std::thread::spawn(move || {
+        // #6185: claim an agent id for this worker BEFORE it can allocate or
+        // enqueue anything, so every pointer it puts in a global queue is
+        // tagged as its own — and so its own drains skip the spawner's work.
+        let worker_agent = crate::agent::enter_worker_agent();
         // Register this thread's root scanners before any allocation can
         // cross a GC trigger (see the parallel_map worker for rationale).
         crate::gc::ensure_gc_initialized();
@@ -1356,15 +1379,28 @@ unsafe fn spawn_impl(closure_val: f64) -> *mut crate::promise::Promise {
 
         match call_result {
             Ok(result) => {
-                // Serialize result for transfer back to main thread
+                // Serialize result for transfer back to the spawning agent.
                 let serialized_result = unsafe { serialize_nanbox_for_thread(result.to_bits()) };
-                queue_thread_result(promise_usize, serialized_result);
+                queue_thread_result(owner_agent, promise_usize, serialized_result);
             }
             Err(_) => {
                 // Thread panicked — resolve with undefined to avoid hanging promise
-                queue_thread_result(promise_usize, SerializedValue::Inline(TAG_UNDEFINED));
+                queue_thread_result(
+                    owner_agent,
+                    promise_usize,
+                    SerializedValue::Inline(TAG_UNDEFINED),
+                );
             }
         }
+
+        // #6185: this worker's arena is about to be unmapped. Drop the shadow
+        // scope first (the result is already serialized into owner-independent
+        // form above), then purge any queue entry still tagged with this agent —
+        // nothing can ever legally settle those, and their pointers are about to
+        // dangle. Must run AFTER the result is queued: that entry is tagged with
+        // `owner_agent`, not `worker_agent`, so it survives the purge.
+        drop(gc_scope);
+        crate::agent::retire_agent(worker_agent);
     });
 
     promise
@@ -1375,7 +1411,11 @@ unsafe fn spawn_impl(closure_val: f64) -> *mut crate::promise::Promise {
 /// Uses the stdlib's PENDING_DEFERRED mechanism. The converter function
 /// runs on the main thread during `js_stdlib_process_pending()`, which
 /// deserializes the value into the main thread's arena.
-fn queue_thread_result(promise_usize: usize, result: SerializedValue) {
+fn queue_thread_result(
+    owner: crate::agent::AgentId,
+    promise_usize: usize,
+    result: SerializedValue,
+) {
     // We need to interact with perry-stdlib's deferred resolution queue.
     // Since perry-runtime cannot depend on perry-stdlib, we use the same
     // pattern as timer resolution: store the result and let the pump pick it up.
@@ -1389,6 +1429,7 @@ fn queue_thread_result(promise_usize: usize, result: SerializedValue) {
             Err(poisoned) => poisoned.into_inner(),
         };
         pending.push(PendingThreadResult {
+            owner,
             promise_ptr: promise_usize,
             result,
         });
@@ -1418,24 +1459,42 @@ pub unsafe fn pin_promise(promise: *mut crate::promise::Promise) {
     (*header).gc_flags |= gc::GC_FLAG_PINNED;
 }
 
-/// Resolve the promise at `promise_usize` with a UTF-8 string on the main
-/// thread. Routes through the same pending-result path `spawn` uses (which
-/// unpins the promise, deserializes the value into the main arena, decrements
-/// the active-job count, and wakes the event loop). Used by `Atomics.waitAsync`.
-pub fn queue_promise_string_result(promise_usize: usize, value: &str) {
+/// Resolve the promise at `promise_usize` with a UTF-8 string on the agent that
+/// owns it. Routes through the same pending-result path `spawn` uses (which
+/// unpins the promise, deserializes the value into that agent's arena,
+/// decrements the active-job count, and wakes the event loop). Used by
+/// `Atomics.waitAsync`.
+///
+/// `owner` must be captured on the thread that CREATED the promise (#6185). The
+/// futex-waiter thread that calls this never runs JS and owns no heap, so it
+/// cannot derive the right agent from itself.
+pub fn queue_promise_string_result(
+    owner: crate::agent::AgentId,
+    promise_usize: usize,
+    value: &str,
+) {
     queue_thread_result(
+        owner,
         promise_usize,
         SerializedValue::String(value.as_bytes().to_vec()),
     );
 }
 
-/// A pending thread result waiting to be resolved on the main thread.
+/// A pending thread result waiting to be resolved on the agent that spawned it.
 struct PendingThreadResult {
+    /// #6185: the agent whose heap `promise_ptr` lives in — captured at spawn
+    /// time from the *spawning* thread, not the worker. Only that agent may
+    /// drain this entry; a worker pumping the global queue would otherwise
+    /// resolve a foreign-heap promise with a pointer into its own arena, which
+    /// is unmapped when it exits.
+    owner: crate::agent::AgentId,
     promise_ptr: usize,
     result: SerializedValue,
 }
 
-// Safety: SerializedValue is Send, usize is Send.
+// Safety: SerializedValue is Send, usize is Send. `promise_ptr` is a raw
+// pointer into `owner`'s arena; the `owner` tag plus the owner-filtered drain
+// in `js_thread_process_pending` is what makes dereferencing it sound.
 unsafe impl Send for PendingThreadResult {}
 
 /// Global queue for pending thread results.
@@ -1452,13 +1511,30 @@ static PENDING_THREAD_RESULTS: std::sync::Mutex<Vec<PendingThreadResult>> =
 /// Number of results processed.
 #[no_mangle]
 pub extern "C" fn js_thread_process_pending() -> i32 {
-    let mut pending = match PENDING_THREAD_RESULTS.lock() {
-        Ok(guard) => guard,
-        Err(poisoned) => poisoned.into_inner(),
+    // #6185: take only the entries THIS agent owns. Every remaining entry names
+    // a promise in another agent's arena; draining it here would resolve a
+    // foreign-heap promise with a value deserialized into our arena (and, once
+    // that agent exits, dereference freed memory). Leave them for their owner —
+    // `retire_agent` purges any whose owner dies first.
+    let mine: Vec<PendingThreadResult> = {
+        let mut pending = match PENDING_THREAD_RESULTS.lock() {
+            Ok(guard) => guard,
+            Err(poisoned) => poisoned.into_inner(),
+        };
+        // Order-preserving partition: results must settle in the order they
+        // were queued (a `swap_remove` filter would reorder them).
+        let (mine, theirs): (Vec<_>, Vec<_>) = std::mem::take(&mut *pending)
+            .into_iter()
+            .partition(|item| crate::agent::owns(item.owner));
+        *pending = theirs;
+        mine
     };
-    let count = pending.len() as i32;
+    let count = mine.len() as i32;
 
-    for item in pending.drain(..) {
+    // The lock is released before we settle anything: `js_promise_resolve` runs
+    // user `.then` callbacks, which can call `spawn` and re-enter
+    // `queue_thread_result` (deadlock on a re-entrant lock of the same Mutex).
+    for item in mine {
         unsafe {
             let promise = item.promise_ptr as *mut crate::promise::Promise;
 
@@ -1493,12 +1569,27 @@ pub extern "C" fn js_thread_has_pending() -> i32 {
     if ACTIVE_THREAD_JOBS.load(Ordering::SeqCst) != 0 {
         return 1;
     }
-    let pending = PENDING_THREAD_RESULTS.lock().unwrap();
-    if pending.is_empty() {
-        0
-    } else {
-        1
-    }
+    // #6185: only entries THIS agent can actually settle count as work keeping
+    // its loop alive. Reporting a foreign entry here would spin the event loop
+    // forever on a result the drain (correctly) refuses to touch.
+    let pending = match PENDING_THREAD_RESULTS.lock() {
+        Ok(guard) => guard,
+        Err(poisoned) => poisoned.into_inner(),
+    };
+    i32::from(pending.iter().any(|item| crate::agent::owns(item.owner)))
+}
+
+/// Drop every queued result owned by `agent`. Called from
+/// `agent::retire_agent` when a worker thread exits: those entries name
+/// promises in an arena that is being unmapped, so no thread can ever settle
+/// them, and leaving them would keep `js_thread_has_pending` honest but the
+/// pointers dangling.
+pub(crate) fn purge_agent_thread_results(agent: crate::agent::AgentId) {
+    let mut pending = match PENDING_THREAD_RESULTS.lock() {
+        Ok(guard) => guard,
+        Err(poisoned) => poisoned.into_inner(),
+    };
+    pending.retain(|item| item.owner != agent);
 }
 
 #[cfg(test)]
diff --git a/crates/perry-runtime/src/timer.rs b/crates/perry-runtime/src/timer.rs
index 94bbcd647e..291d1fc599 100644
--- a/crates/perry-runtime/src/timer.rs
+++ b/crates/perry-runtime/src/timer.rs
@@ -22,6 +22,9 @@ extern "C" {
 
 /// A scheduled timer
 struct Timer {
+    /// #6185: agent whose heap `promise` lives in; only it (or a pump acting for
+    /// it — see `crate::agent`) may fire this timer.
+    owner: crate::agent::AgentId,
     /// When this timer should fire
     deadline: Instant,
     /// The promise to resolve when the timer fires
@@ -32,7 +35,10 @@ struct Timer {
     has_ref: bool,
 }
 
-// SAFETY: Promise pointers are only accessed from the pump thread
+// SAFETY: `promise` points into `owner`'s arena. The pre-#6185 claim here was
+// "only accessed from the pump thread", which nothing enforced — any thread
+// running the await loop drained this queue. The `owner` tag plus the
+// owner-filtered tick is what makes that claim true.
 unsafe impl Send for Timer {}
 
 // Global timer queues (Mutex-protected for cross-thread access)
@@ -87,6 +93,8 @@ fn schedule_promise_timer(delay_ms: f64, value: f64, has_ref: bool) -> *mut Prom
     let deadline = Instant::now() + delay;
 
     TIMER_QUEUE.lock().unwrap().push(Timer {
+        // #6185: tag with the scheduling agent — only it may fire this.
+        owner: crate::agent::current_agent(),
         deadline,
         promise,
         value,
@@ -96,14 +104,6 @@ fn schedule_promise_timer(delay_ms: f64, value: f64, has_ref: bool) -> *mut Prom
     promise
 }
 
-fn has_refed_promise_timer() -> bool {
-    TIMER_QUEUE
-        .lock()
-        .unwrap()
-        .iter()
-        .any(|timer| timer.has_ref)
-}
-
 fn timer_has_ref_state(id: i64) -> bool {
     TIMER_REF_STATES
         .lock()
@@ -113,22 +113,6 @@ fn timer_has_ref_state(id: i64) -> bool {
         .unwrap_or(true)
 }
 
-fn has_refed_callback_timer() -> bool {
-    CALLBACK_TIMERS
-        .lock()
-        .unwrap()
-        .iter()
-        .any(|timer| !timer.cleared && timer_has_ref_state(timer.id))
-}
-
-fn has_refed_interval_timer() -> bool {
-    INTERVAL_TIMERS
-        .lock()
-        .unwrap()
-        .iter()
-        .any(|timer| !timer.cleared && timer_has_ref_state(timer.id))
-}
-
 fn other_event_sources_keep_loop_alive() -> bool {
     has_refed_callback_timer()
         || has_refed_interval_timer()
@@ -215,7 +199,15 @@ pub extern "C" fn js_timer_tick() -> i32 {
         drain_expired_timers(
             &mut queue,
             |_| false,
-            |timer| timer.deadline <= now && (timer.has_ref || allow_unref),
+            // #6185: never fire another agent's timer — its promise and value are
+            // pointers into that agent's arena. A non-owned entry fails the
+            // predicate, so the partition returns it to the queue for its real
+            // owner rather than firing or dropping it.
+            |timer| {
+                crate::agent::owns(timer.owner)
+                    && timer.deadline <= now
+                    && (timer.has_ref || allow_unref)
+            },
         )
     };
     // #6287: fire the batch in deadline order, not creation order — a 5 ms
@@ -266,7 +258,7 @@ pub extern "C" fn js_timer_next_deadline() -> f64 {
         .lock()
         .unwrap()
         .iter()
-        .filter(|t| t.has_ref || allow_unref)
+        .filter(|t| (t.has_ref || allow_unref) && crate::agent::owns(t.owner))
         .map(|t| {
             if t.deadline <= now {
                 0.0
@@ -320,9 +312,15 @@ struct CallbackTimer {
     trigger_async_id: u64,
     /// Whether this timer has been cleared
     cleared: bool,
+    /// #6185: agent whose heap `callback` (and any pointer-valued `args`) live
+    /// in. Only that agent — or a pump acting for it, e.g. Android's UI thread
+    /// for the primary agent — may fire it.
+    owner: crate::agent::AgentId,
 }
 
-// SAFETY: closure pointers point to global compiled code data
+// SAFETY: the closure POINTER targets global compiled code, but the closure
+// OBJECT and any NaN-boxed `args` live in `owner`'s arena; the owner tag plus
+// the owner-filtered tick is what makes firing them sound.
 unsafe impl Send for CallbackTimer {}
 
 pub const MOCK_TIMERS_API_DATE: u32 = 1 << 0;
@@ -388,8 +386,12 @@ static NEXT_TIMER_ID: Mutex<i64> = Mutex::new(1);
 // #6084: the bounded ref-state registry lives in a submodule to keep this file
 // under the 2000-line lint cap.
 mod gc_scan;
+mod ownership;
 mod ref_states;
 
+pub(crate) use ownership::purge_agent_timers;
+use ownership::{has_refed_callback_timer, has_refed_interval_timer, has_refed_promise_timer};
+
 pub(crate) use gc_scan::{new_timer_root_scan_state, scan_timer_roots_mut_step};
 use ref_states::{TimerRefStates, TIMER_REF_STATES_CAP};
 
@@ -1047,6 +1049,8 @@ fn schedule_callback_timer(
         async_id: ids.async_id,
         trigger_async_id: ids.trigger_async_id,
         cleared: false,
+        // #6185: the scheduling agent owns the callback closure + args.
+        owner: crate::agent::current_agent(),
     });
     set_timer_ref_state(id, true);
 
@@ -1126,8 +1130,17 @@ pub extern "C" fn js_callback_timer_tick() -> i32 {
         let mut queue = CALLBACK_TIMERS.lock().unwrap();
         drain_expired_timers(
             &mut queue,
+            // Dropping a cleared timer is safe regardless of owner: nothing here
+            // dereferences its callback, we just release the entry.
             |timer| timer.cleared,
-            |timer| timer.deadline <= now && (timer_has_ref_state(timer.id) || allow_unref),
+            // #6185: only ever call back into OUR OWN agent's heap. Firing a
+            // foreign agent's closure here would run main-heap JS on a worker (or
+            // vice versa) and allocate the results in the wrong arena.
+            |timer| {
+                crate::agent::owns(timer.owner)
+                    && timer.deadline <= now
+                    && (timer_has_ref_state(timer.id) || allow_unref)
+            },
         )
     };
     // #6287: timers phase (by deadline) before check phase (FIFO immediates).
@@ -1223,17 +1236,22 @@ pub extern "C" fn js_callback_timer_has_pending() -> i32 {
 }
 
 pub fn active_timeout_resource_count() -> usize {
+    // #6185: a timer another agent scheduled is not this agent's active handle.
     let callback_count = CALLBACK_TIMERS
         .lock()
         .unwrap()
         .iter()
-        .filter(|timer| !timer.cleared && timer.kind == CallbackTimerKind::Timeout)
+        .filter(|timer| {
+            !timer.cleared
+                && timer.kind == CallbackTimerKind::Timeout
+                && crate::agent::owns(timer.owner)
+        })
         .count();
     let interval_count = INTERVAL_TIMERS
         .lock()
         .unwrap()
         .iter()
-        .filter(|timer| !timer.cleared)
+        .filter(|timer| !timer.cleared && crate::agent::owns(timer.owner))
         .count();
     let mock_count = {
         let state = MOCK_TIMERS.lock().unwrap();
@@ -1265,7 +1283,9 @@ pub extern "C" fn js_callback_timer_next_deadline() -> f64 {
         .lock()
         .unwrap()
         .iter()
-        .filter(|t| !t.cleared && (timer_has_ref_state(t.id) || allow_unref))
+        .filter(|t| {
+            !t.cleared && crate::agent::owns(t.owner) && (timer_has_ref_state(t.id) || allow_unref)
+        })
         .map(|t| {
             if t.deadline <= now {
                 0.0
@@ -1386,9 +1406,12 @@ struct IntervalTimer {
     context: crate::async_context::AsyncContextSnapshot,
     /// Whether this interval has been cleared
     cleared: bool,
+    /// #6185: agent that owns `callback` / `args`. See `CallbackTimer::owner`.
+    owner: crate::agent::AgentId,
 }
 
-// SAFETY: closure pointers point to global compiled code data
+// SAFETY: see `CallbackTimer` — the owner tag plus owner-filtered ticking is
+// what makes the cross-thread pointers here sound.
 unsafe impl Send for IntervalTimer {}
 
 static INTERVAL_TIMERS: Mutex<Vec<IntervalTimer>> = Mutex::new(Vec::new());
@@ -1420,6 +1443,8 @@ fn schedule_interval_timer(callback: i64, interval_ms: f64, args: Vec<f64>) -> i
         args,
         context: crate::async_context::capture_context(),
         cleared: false,
+        // #6185: the scheduling agent owns the callback closure + args.
+        owner: crate::agent::current_agent(),
     });
     set_timer_ref_state(id, true);
 
@@ -1494,7 +1519,10 @@ pub extern "C" fn js_interval_timer_tick() -> i32 {
         let mut callbacks = Vec::new();
 
         for timer in timers.iter_mut() {
+            // #6185: never fire a foreign agent's interval callback — the closure
+            // and its args live in that agent's arena.
             if !timer.cleared
+                && crate::agent::owns(timer.owner)
                 && timer.next_deadline <= now
                 && (timer_has_ref_state(timer.id) || allow_unref)
             {
@@ -1575,7 +1603,9 @@ pub extern "C" fn js_interval_timer_next_deadline() -> f64 {
         .lock()
         .unwrap()
         .iter()
-        .filter(|t| !t.cleared && (timer_has_ref_state(t.id) || allow_unref))
+        .filter(|t| {
+            !t.cleared && crate::agent::owns(t.owner) && (timer_has_ref_state(t.id) || allow_unref)
+        })
         .map(|t| {
             if t.next_deadline <= now {
                 0.0
@@ -1593,11 +1623,18 @@ pub fn scan_timer_roots(mark: &mut dyn FnMut(f64)) {
     scan_timer_roots_mut(&mut visitor);
 }
 
+/// #6185: scan ONLY this agent's timers. Every pointer in these queues belongs
+/// to the arena of the agent that scheduled the timer. A GC cycle on agent A
+/// that walked agent B's entries would mark through B's heap (racing B's
+/// collector on the same GcHeader bits) and, on an evacuating cycle, REWRITE B's
+/// slots to forwarding addresses in A's arena — corrupting a heap it does not
+/// own. Foreign timers are rooted by their own agent's collector, the only one
+/// that can see their arena. `gc_scan.rs` applies the same rule incrementally.
 pub fn scan_timer_roots_mut(visitor: &mut crate::gc::RuntimeRootVisitor<'_>) {
     // Scan promise-based timers
     {
         let mut q = TIMER_QUEUE.lock().unwrap();
-        for timer in q.iter_mut() {
+        for timer in q.iter_mut().filter(|t| crate::agent::owns(t.owner)) {
             visitor.visit_raw_mut_ptr_slot(&mut timer.promise);
             visitor.visit_nanbox_f64_slot(&mut timer.value);
         }
@@ -1606,7 +1643,7 @@ pub fn scan_timer_roots_mut(visitor: &mut crate::gc::RuntimeRootVisitor<'_>) {
     // Scan callback timers (closure pointers stored as i64)
     {
         let mut q = CALLBACK_TIMERS.lock().unwrap();
-        for timer in q.iter_mut() {
+        for timer in q.iter_mut().filter(|t| crate::agent::owns(t.owner)) {
             if !timer.cleared && timer.callback != 0 {
                 visitor.visit_i64_slot(&mut timer.callback);
             }
@@ -1623,7 +1660,7 @@ pub fn scan_timer_roots_mut(visitor: &mut crate::gc::RuntimeRootVisitor<'_>) {
     // Scan interval timers
     {
         let mut q = INTERVAL_TIMERS.lock().unwrap();
-        for timer in q.iter_mut() {
+        for timer in q.iter_mut().filter(|t| crate::agent::owns(t.owner)) {
             if !timer.cleared && timer.callback != 0 {
                 visitor.visit_i64_slot(&mut timer.callback);
             }
@@ -1717,12 +1754,16 @@ pub(crate) fn test_seed_timer_scanner_roots(
     let context = crate::async_context::test_snapshot_with_store(context_store);
     let deadline = Instant::now() + Duration::from_secs(86_400);
     TIMER_QUEUE.lock().unwrap().push(Timer {
+        // #6185: test scaffolding runs on the primary agent.
+        owner: crate::agent::current_agent(),
         deadline,
         promise,
         value,
         has_ref: true,
     });
     CALLBACK_TIMERS.lock().unwrap().push(CallbackTimer {
+        // #6185: test scaffolding runs on the primary agent.
+        owner: crate::agent::current_agent(),
         id: TEST_CALLBACK_TIMER_ID,
         kind: CallbackTimerKind::Timeout,
         deadline,
@@ -1735,6 +1776,8 @@ pub(crate) fn test_seed_timer_scanner_roots(
         cleared: false,
     });
     INTERVAL_TIMERS.lock().unwrap().push(IntervalTimer {
+        // #6185: test scaffolding runs on the primary agent.
+        owner: crate::agent::current_agent(),
         id: TEST_INTERVAL_TIMER_ID,
         callback,
         interval_ms: 86_400_000,
@@ -1752,6 +1795,8 @@ pub(crate) fn test_seed_many_timeout_roots(values: &[f64]) {
     q.clear();
     for &value in values {
         q.push(Timer {
+            // #6185: test scaffolding runs on the primary agent.
+            owner: crate::agent::current_agent(),
             deadline,
             promise: std::ptr::null_mut(),
             value,
@@ -1878,6 +1923,8 @@ mod expired_batch_order_tests {
 
     fn timer(id: i64, kind: CallbackTimerKind, base: Instant, delay_ms: u64) -> CallbackTimer {
         CallbackTimer {
+            // #6185: test scaffolding runs on the primary agent.
+            owner: crate::agent::current_agent(),
             id,
             kind,
             deadline: base + Duration::from_millis(delay_ms),
diff --git a/crates/perry-runtime/src/timer/gc_scan.rs b/crates/perry-runtime/src/timer/gc_scan.rs
index 8bb628089f..6e6dc3bb27 100644
--- a/crates/perry-runtime/src/timer/gc_scan.rs
+++ b/crates/perry-runtime/src/timer/gc_scan.rs
@@ -53,6 +53,15 @@ fn scan_timeout_timers_step(
     let mut q = TIMER_QUEUE.lock().unwrap();
     while state.index < q.len() {
         let timer = &mut q[state.index];
+        // #6185: the budgeted/incremental scan obeys the same ownership rule as
+        // the stop-the-world one — never mark (and, on an evacuating cycle, never
+        // rewrite) a slot in another agent's arena. Skip the entry wholesale,
+        // advancing the scan state exactly as if it had been fully scanned.
+        if !crate::agent::owns(timer.owner) {
+            state.index += 1;
+            state.finish_timer();
+            continue;
+        }
         while state.slot < 2 {
             if !consume_timer_root_work(remaining) {
                 return false;
@@ -78,6 +87,15 @@ fn scan_callback_timers_step(
     let mut q = CALLBACK_TIMERS.lock().unwrap();
     while state.index < q.len() {
         let timer = &mut q[state.index];
+        // #6185: the budgeted/incremental scan obeys the same ownership rule as
+        // the stop-the-world one — never mark (and, on an evacuating cycle, never
+        // rewrite) a slot in another agent's arena. Skip the entry wholesale,
+        // advancing the scan state exactly as if it had been fully scanned.
+        if !crate::agent::owns(timer.owner) {
+            state.index += 1;
+            state.finish_timer();
+            continue;
+        }
         if state.slot == 0 {
             if !consume_timer_root_work(remaining) {
                 return false;
@@ -137,6 +155,15 @@ fn scan_interval_timers_step(
     let mut q = INTERVAL_TIMERS.lock().unwrap();
     while state.index < q.len() {
         let timer = &mut q[state.index];
+        // #6185: the budgeted/incremental scan obeys the same ownership rule as
+        // the stop-the-world one — never mark (and, on an evacuating cycle, never
+        // rewrite) a slot in another agent's arena. Skip the entry wholesale,
+        // advancing the scan state exactly as if it had been fully scanned.
+        if !crate::agent::owns(timer.owner) {
+            state.index += 1;
+            state.finish_timer();
+            continue;
+        }
         if state.slot == 0 {
             if !consume_timer_root_work(remaining) {
                 return false;
diff --git a/crates/perry-runtime/src/timer/ownership.rs b/crates/perry-runtime/src/timer/ownership.rs
new file mode 100644
index 0000000000..ea882b2a1a
--- /dev/null
+++ b/crates/perry-runtime/src/timer/ownership.rs
@@ -0,0 +1,57 @@
+//! Agent-ownership bookkeeping for the timer queues (#6185 Tier 2).
+//!
+//! The queues in `timer.rs` are process-global, but every pointer they hold —
+//! promise, callback closure, NaN-boxed arg — belongs to the arena of the agent
+//! that scheduled the timer. `timer.rs` enforces that at every read (tick,
+//! next-deadline, liveness, active-handle count) and `gc_scan.rs` enforces it in
+//! the collector. This module owns the two pieces that are purely about
+//! ownership: per-agent event-loop liveness, and what happens to an agent's
+//! timers when the agent itself goes away.
+
+use super::{timer_has_ref_state, CALLBACK_TIMERS, INTERVAL_TIMERS, TIMER_QUEUE};
+
+// ── Per-agent event-loop liveness ────────────────────────────────────────────
+//
+// #6185: a timer owned by another agent can never be fired by this one, so it
+// must not keep this agent's event loop spinning. Pre-fix these counted every
+// timer on the process-global queues, so one agent's pending work kept every
+// other agent's loop alive.
+
+pub(super) fn has_refed_promise_timer() -> bool {
+    TIMER_QUEUE
+        .lock()
+        .unwrap()
+        .iter()
+        .any(|timer| timer.has_ref && crate::agent::owns(timer.owner))
+}
+
+pub(super) fn has_refed_callback_timer() -> bool {
+    CALLBACK_TIMERS.lock().unwrap().iter().any(|timer| {
+        !timer.cleared && crate::agent::owns(timer.owner) && timer_has_ref_state(timer.id)
+    })
+}
+
+pub(super) fn has_refed_interval_timer() -> bool {
+    INTERVAL_TIMERS.lock().unwrap().iter().any(|timer| {
+        !timer.cleared && crate::agent::owns(timer.owner) && timer_has_ref_state(timer.id)
+    })
+}
+
+/// Drop every timer owned by `agent`. Called from `crate::agent::retire_agent`
+/// when a `perry/thread` worker exits.
+///
+/// A timer scheduled inside a worker can never legally fire: the worker has no
+/// event loop of its own (Tier 1, #6276, rejects `await` in a worker body, so it
+/// cannot pump), and no other agent may run its callback — the closure lives in
+/// the worker's arena, which is unmapped at exit. Pre-#6185 such a timer *did*
+/// fire, on whichever thread pumped next, dereferencing that freed arena.
+///
+/// Dropping the entry is therefore the honest end state, not a loss of work: it
+/// is what "the thread ended before the timer came due" already meant. Purging
+/// is safe precisely because nothing else can own these entries — no other
+/// thread may dereference pointers into this agent's arena.
+pub(crate) fn purge_agent_timers(agent: crate::agent::AgentId) {
+    TIMER_QUEUE.lock().unwrap().retain(|t| t.owner != agent);
+    CALLBACK_TIMERS.lock().unwrap().retain(|t| t.owner != agent);
+    INTERVAL_TIMERS.lock().unwrap().retain(|t| t.owner != agent);
+}
diff --git a/crates/perry/tests/issue_6185_thread_owner_tagged_dispatch.rs b/crates/perry/tests/issue_6185_thread_owner_tagged_dispatch.rs
new file mode 100644
index 0000000000..0ba299d079
--- /dev/null
+++ b/crates/perry/tests/issue_6185_thread_owner_tagged_dispatch.rs
@@ -0,0 +1,252 @@
+//! #6185 Tier 2 (end-to-end): owner-tagged cross-thread dispatch must not
+//! disturb any of the shipped `perry/thread` / timer behavior.
+//!
+//! Tier 2 tags every entry on the three process-global queues
+//! (`PENDING_THREAD_RESULTS`, `TIMER_QUEUE` / `CALLBACK_TIMERS` /
+//! `INTERVAL_TIMERS`) with the agent that owns the heap its pointers live in,
+//! and makes every drain skip entries it does not own. The unsound drains it
+//! removes are not directly expressible in TypeScript anymore — Tier 1 (#6276)
+//! rejects the `await`-inside-a-worker that used to reach them — so the property
+//! under test here is the *other* half: the queues still deliver everything they
+//! are supposed to, on the thread that owns them.
+//!
+//! The owner-filtering itself (a worker neither firing nor eating a foreign
+//! agent's timer; a retired worker's timers being purged) is asserted directly
+//! against the runtime in `perry-runtime/src/agent_dispatch_tests.rs`.
+
+use std::path::PathBuf;
+use std::process::{Command, Stdio};
+use std::time::{Duration, Instant};
+
+/// The failure mode for most of these is a HANG, not a wrong answer: an
+/// owner-tag regression files a result under the wrong agent, so the promise it
+/// belongs to never settles and the program spins its event loop forever.
+/// `cargo test` has no per-test timeout and `Command::output()` blocks
+/// indefinitely, so an unbounded run would stall the whole suite with no
+/// diagnostic. Bound it and report the hang as a normal test failure instead.
+const RUN_TIMEOUT: Duration = Duration::from_secs(60);
+
+fn perry_bin() -> PathBuf {
+    PathBuf::from(env!("CARGO_BIN_EXE_perry"))
+}
+
+fn compile_and_run(source: &str) -> String {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let entry = dir.path().join("main.ts");
+    let output = dir.path().join("main_bin");
+    std::fs::write(&entry, source).expect("write entry");
+
+    let compile = Command::new(perry_bin())
+        .current_dir(dir.path())
+        .arg("compile")
+        .arg(&entry)
+        .arg("-o")
+        .arg(&output)
+        .output()
+        .expect("run perry compile");
+    assert!(
+        compile.status.success(),
+        "perry compile failed\nstdout:\n{}\nstderr:\n{}",
+        String::from_utf8_lossy(&compile.stdout),
+        String::from_utf8_lossy(&compile.stderr)
+    );
+
+    // Piped stdio + poll, so a program that never settles is killed and
+    // reported rather than blocking the harness forever. These programs print a
+    // few dozen bytes, so they cannot fill the pipe buffer while we poll.
+    let mut child = Command::new(&output)
+        .current_dir(dir.path())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .expect("spawn compiled binary");
+
+    let deadline = Instant::now() + RUN_TIMEOUT;
+    loop {
+        match child.try_wait().expect("poll compiled binary") {
+            Some(_) => break,
+            None if Instant::now() >= deadline => {
+                let _ = child.kill();
+                let out = child.wait_with_output().expect("reap timed-out binary");
+                panic!(
+                    "compiled binary did not exit within {:?} — a promise never settled \
+                     (owner-tag regression: work filed under an agent that never drains it)\
+                     \nstdout so far:\n{}\nstderr so far:\n{}",
+                    RUN_TIMEOUT,
+                    String::from_utf8_lossy(&out.stdout),
+                    String::from_utf8_lossy(&out.stderr)
+                );
+            }
+            None => std::thread::sleep(Duration::from_millis(20)),
+        }
+    }
+
+    let run = child.wait_with_output().expect("reap compiled binary");
+    assert!(
+        run.status.success(),
+        "compiled binary failed\nstatus: {:?}\nstdout:\n{}\nstderr:\n{}",
+        run.status,
+        String::from_utf8_lossy(&run.stdout),
+        String::from_utf8_lossy(&run.stderr)
+    );
+    String::from_utf8_lossy(&run.stdout).into_owned()
+}
+
+/// `spawn`'s completion still resolves on the agent that spawned it. This is the
+/// queue whose drain was previously unconditional: the result is tagged with the
+/// *spawning* agent (captured on the spawning thread, not inside the worker), so
+/// the main thread — and only the main thread — settles the promise.
+#[test]
+fn spawn_result_still_resolves_on_the_spawning_agent() {
+    let out = compile_and_run(
+        r#"
+import { spawn } from "perry/thread";
+async function main(): Promise<void> {
+  const result = await spawn(() => {
+    let sum = 0;
+    for (let i = 1; i <= 100; i++) sum += i;
+    return sum;
+  });
+  console.log("spawn:" + result);
+}
+main();
+"#,
+    );
+    assert_eq!(out.trim(), "spawn:5050");
+}
+
+/// Many concurrent `spawn`s settle in the right order with the right values —
+/// the owner-filtered drain is an order-preserving partition, not a
+/// `swap_remove` filter (which would shuffle the settle order).
+#[test]
+fn concurrent_spawn_results_all_settle_in_order() {
+    let out = compile_and_run(
+        r#"
+import { spawn } from "perry/thread";
+async function main(): Promise<void> {
+  const jobs = [1, 2, 3, 4, 5, 6, 7, 8].map((n) =>
+    spawn(() => {
+      let acc = 0;
+      for (let i = 0; i < 1000; i++) acc += n;
+      return acc;
+    })
+  );
+  const results = await Promise.all(jobs);
+  console.log(results.join(","));
+}
+main();
+"#,
+    );
+    assert_eq!(out.trim(), "1000,2000,3000,4000,5000,6000,7000,8000");
+}
+
+/// `parallelMap` / `parallelFilter` workers each claim (and retire) their own
+/// agent. Their results must be unaffected and order-preserved.
+#[test]
+fn parallel_map_and_filter_still_work_across_worker_agents() {
+    let out = compile_and_run(
+        r#"
+import { parallelMap, parallelFilter } from "perry/thread";
+const nums = [1, 2, 3, 4, 5, 6, 7, 8];
+const doubled = parallelMap(nums, (x: number) => x * 2);
+const evens = parallelFilter(nums, (x: number) => x % 2 === 0);
+console.log("map:" + doubled.join(","));
+console.log("filter:" + evens.join(","));
+"#,
+    );
+    assert_eq!(out.trim(), "map:2,4,6,8,10,12,14,16\nfilter:2,4,6,8");
+}
+
+/// The main agent's timers, intervals, immediates and microtasks all still fire,
+/// in spec order, now that every tick is owner-filtered. A regression to a raw
+/// `ThreadId` comparison (or a mis-tagged primary agent) shows up here as
+/// missing output / a hang.
+#[test]
+fn main_agent_timers_intervals_and_microtasks_all_still_fire_in_order() {
+    let out = compile_and_run(
+        r#"
+const order: string[] = [];
+setTimeout(() => order.push("t100"), 100);
+setTimeout(() => order.push("t0"), 0);
+setImmediate(() => order.push("immediate"));
+Promise.resolve().then(() => order.push("microtask"));
+queueMicrotask(() => order.push("qmt"));
+let n = 0;
+const iv = setInterval(() => {
+  n++;
+  order.push("interval" + n);
+  if (n === 3) clearInterval(iv);
+}, 20);
+const cancelled = setTimeout(() => order.push("SHOULD-NOT-FIRE"), 30);
+clearTimeout(cancelled);
+await new Promise<void>((r) => setTimeout(r, 250));
+console.log(order.join(","));
+"#,
+    );
+    // Same ordering `node --experimental-strip-types` produces for this program.
+    assert_eq!(
+        out.trim(),
+        "microtask,qmt,t0,immediate,interval1,interval2,interval3,t100"
+    );
+}
+
+/// Timers scheduled while worker agents are alive still belong to — and fire on
+/// — the main agent. The worker's own tick (had it run one) would skip them, and
+/// retiring the worker must not purge them: `retire_agent` keys on the owner tag,
+/// so only the worker's own entries go.
+#[test]
+fn worker_lifetime_does_not_disturb_the_main_agents_timers() {
+    let out = compile_and_run(
+        r#"
+import { spawn, parallelMap } from "perry/thread";
+async function main(): Promise<void> {
+  const fired: string[] = [];
+  setTimeout(() => fired.push("before-worker"), 30);
+
+  // Workers come and go (each claims an agent, then retires it) while the
+  // main agent's timers are pending.
+  const doubled = parallelMap([1, 2, 3], (x: number) => x * 2);
+  const sum = await spawn(() => 7 * 6);
+
+  setTimeout(() => fired.push("after-worker"), 60);
+  await new Promise<void>((r) => setTimeout(r, 200));
+
+  console.log("map:" + doubled.join(","));
+  console.log("spawn:" + sum);
+  console.log("timers:" + fired.join(","));
+}
+main();
+"#,
+    );
+    assert_eq!(
+        out.trim(),
+        "map:2,4,6\nspawn:42\ntimers:before-worker,after-worker"
+    );
+}
+
+/// `Atomics.waitAsync` resolves through the same pending-result queue as
+/// `spawn`, but its promise is created by the *calling* agent while the futex
+/// waiter thread (which owns no heap and runs no JS) queues the result. The
+/// owner tag must be captured at the call site, or the result is filed under the
+/// wrong agent and the promise never settles (the test hangs).
+#[test]
+fn atomics_wait_async_resolves_through_the_owner_tagged_queue() {
+    let out = compile_and_run(
+        r#"
+import { spawn } from "perry/thread";
+const sab = new SharedArrayBuffer(4);
+const a = new Int32Array(sab);
+const r = Atomics.waitAsync(a, 0, 0, 10000) as { async: boolean; value: Promise<string> };
+console.log("async:" + r.async);
+spawn(() => {
+  const view = new Int32Array(sab);
+  const nap = new Int32Array(new SharedArrayBuffer(4));
+  Atomics.wait(nap, 0, 0, 30);
+  return Atomics.notify(view, 0);
+});
+const result = await r.value;
+console.log("result:" + result);
+"#,
+    );
+    assert_eq!(out.trim(), "async:true\nresult:ok");
+}