diff --git a/.gitignore b/.gitignore
index 24a239c7..90242bcb 100644
--- a/.gitignore
+++ b/.gitignore
@@ -55,3 +55,6 @@ xcuserdata/
 !/docs/_includes/
 !/docs/assets/
 !/docs/ramps.json
+
+# Per-worktree DerivedData (docs: .claude/skills/pr-gate-report)
+.derived/
diff --git a/GraphcodeKit/Sources/Deadline.swift b/GraphcodeKit/Sources/Deadline.swift
new file mode 100644
index 00000000..f347a229
--- /dev/null
+++ b/GraphcodeKit/Sources/Deadline.swift
@@ -0,0 +1,60 @@
+import Foundation
+
+/// Runs `operation` with a deadline, answering `nil` when it has not finished in time.
+///
+/// The slower of the two is **abandoned**, not cancelled and waited for, and that is the
+/// whole point. A structured `withTaskGroup` race cannot bound anything here: the group
+/// waits for every child before it returns, so a child that never finishes takes the
+/// timeout down with it. The reads this exists for are exactly that kind — a presence
+/// probe reaches `PTYProcessSession.waitCollectingOutput`, which ends only when the
+/// child's `terminationHandler` closes the stream, and `ssh`'s `ConnectTimeout` bounds
+/// the *connect*, never a command already running on a host that has gone away
+/// (`RemoteEnsureGate` documents the same wedge from the other side).
+///
+/// Cancelling the loser is therefore hygiene rather than the mechanism: a task blocked
+/// on a pipe that will never close ignores it. What makes this safe is that nothing
+/// awaits the abandoned task, so it costs one suspended task until its subprocess is
+/// reaped, and the caller is already gone.
+func withDeadline<T: Sendable>(
+  _ deadline: Duration, _ operation: @escaping @Sendable () async -> T
+) async -> T? {
+  let relay = DeadlineRelay<T>()
+  let work = Task.detached { await relay.settle(await operation()) }
+  let timer = Task.detached {
+    try? await Task.sleep(for: deadline)
+    await relay.settle(nil)
+  }
+  let answer = await relay.wait()
+  work.cancel()
+  timer.cancel()
+  return answer
+}
+
+/// Whichever of the two tasks arrives first wins; the other's answer is dropped rather
+/// than resuming a continuation twice.
+private actor DeadlineRelay<T: Sendable> {
+  private var answer: T??
+  private var waiter: CheckedContinuation<T?, Never>?
+
+  func settle(_ value: T?) {
+    guard answer == nil else { return }
+    answer = .some(value)
+    guard let waiter else { return }
+    self.waiter = nil
+    waiter.resume(returning: value)
+  }
+
+  func wait() async -> T? {
+    if let answer { return answer }
+    return await withCheckedContinuation { self.waiter = $0 }
+  }
+}
+
+extension Duration {
+  /// The same span as `Date` arithmetic wants it. `Duration` is what the concurrency
+  /// APIs take and `TimeInterval` is what a lease compares against; this is the one
+  /// conversion between them.
+  var timeInterval: TimeInterval {
+    Double(components.seconds) + Double(components.attoseconds) / 1e18
+  }
+}
diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift
index d9dc444c..92231dd4 100644
--- a/GraphcodeKit/Sources/GraphStore.swift
+++ b/GraphcodeKit/Sources/GraphStore.swift
@@ -182,6 +182,10 @@ public actor GraphStore {
     /// still queued, and lets a wake for a post the reader has since read go unsent.
     /// `nil` for a message a peer or a human sent.
     let watchedPostID: Int?
+    /// Whether the target's memory log already carries this message. Written the first
+    /// time the item is actually put back rather than when it was queued — see
+    /// `staged(_:)`.
+    var recorded: Bool = false
   }
 
   private var pendingFollowUps: [PendingFollowUp] = []
@@ -190,7 +194,49 @@ public actor GraphStore {
   /// the first was suspended delivered the same items again and, on finishing, wrote
   /// its own idea of what remained over the first's — dropping whatever had been
   /// queued in between (issue #304: duplicates, lost mail, and out-of-order delivery).
-  private var isDrainingFollowUps = false
+  /// A drain in flight, held as a **lease** rather than a flag: the moment it was taken,
+  /// which is also proof of *which* drain holds it.
+  ///
+  /// A bare boolean was enough while the only unbounded await under it was the presence
+  /// read, which now has a deadline. It is not enough for the class: `deliverToSession`
+  /// is the same `PTYProcessSession` chain with no deadline of its own — `zmx ls` to
+  /// check the session exists, a write per chunk, and `sendRemote` over `ssh` for a
+  /// remote loop — so a hang there would hold a bare flag for the life of the daemon
+  /// exactly as the presence read did, and every loop in the project would stop
+  /// receiving mail with nothing logged (measured: 364s, zero errors). `RemoteEnsureGate`
+  /// rejects the plain flag for this same chain and for this same reason.
+  ///
+  /// So the guard expires. A drain that outlives its lease is a wedge by definition, and
+  /// the next one says so in the daemon log and takes over. Taking over cannot duplicate
+  /// anything — the batch was taken and cleared in one actor step, so the successor
+  /// finds only what was queued after it — and if the wedged drain ever returns, its
+  /// `defer` folds its own remainder back on. Order can slip in that recovery; a queue
+  /// that moves again beats one frozen for ever.
+  private var drainLease: Date?
+
+  /// How long a presence read may take before the store stops waiting on it.
+  ///
+  /// Every reading in this file runs through `presenceReading(of:)`, and nothing in the
+  /// chain below it has a deadline of its own: `onReadPresence` reaches
+  /// `PTYProcessSession.waitCollectingOutput`, which ends only when the probe's
+  /// `terminationHandler` closes the stream, and `ssh`'s `ConnectTimeout=10` bounds the
+  /// connect rather than a command left hanging on a host that has gone away. An `await`
+  /// that never returns held the drain's guard for the life of the daemon, which
+  /// froze staged delivery for *every* loop in the project — and a frozen queue and an
+  /// empty one report exactly the same thing from outside (issue #311).
+  ///
+  /// Longer than any healthy read: a remote probe is three attempts at
+  /// `ConnectTimeout=10` with 1s and 2s of backoff between them (`RemoteEnsureGate`),
+  /// so a live-but-slow host still answers inside this. Short enough that a wedge costs
+  /// one poll, not the process.
+  private let presenceReadDeadline: Duration
+
+  /// How long a drain may hold the queue before another is allowed to take over. Far
+  /// longer than any healthy pass (milliseconds) and longer than a pass that meets
+  /// several wedged presence reads, each of which is bounded by `presenceReadDeadline`;
+  /// short enough that a hang costs minutes rather than the life of the process. The
+  /// 300s here is `RemoteEnsureGate.leaseDuration`, arrived at for the same chain.
+  private let drainLeaseDuration: Duration
 
   /// A poller holds `self` weakly, so a store going away already stops it *doing*
   /// anything — but the task itself keeps sleeping in its loop forever. Harmless for
@@ -263,6 +309,8 @@ public actor GraphStore {
     onMailroomEnabled: (@Sendable () -> Bool)? = nil,
     goalCache: GoalEvaluationCache? = nil,
     recurrence: RecurrenceSink? = nil,
+    presenceReadDeadline: Duration = .seconds(45),
+    drainLeaseDuration: Duration = .seconds(300),
     subGraphDepth: Int = 0
   ) {
     self.graph = graph
@@ -294,6 +342,25 @@ public actor GraphStore {
     self.onMailroomEnabled = onMailroomEnabled
     self.goalCache = goalCache ?? GoalEvaluationCache()
     self.recurrence = recurrence
+    self.presenceReadDeadline = presenceReadDeadline
+    self.drainLeaseDuration = drainLeaseDuration
+  }
+
+  /// The store's one way to ask what a session is doing, and the only place the answer
+  /// is bounded. `nil` means there is no reader wired at all — every caller then falls
+  /// back to whatever the graph already believes.
+  ///
+  /// A read that runs out of time is `.unknown`, never a state. That is the distinction
+  /// issue #286 established when a `zmx` probe that could not run was being read as
+  /// `.absent`: a probe that did not complete says nothing about the session, so a
+  /// caller must not turn it into a verdict. Here it means a staged message stays
+  /// staged and is tried again, rather than being delivered blindly into a session that
+  /// may be mid-turn or held back for ever as if the target were busy.
+  private func presenceReading(of node: LoopNode) async -> PresenceReading? {
+    guard let onReadPresence else { return nil }
+    let path = graph.project.path
+    return await withDeadline(presenceReadDeadline) { await onReadPresence(node, path) }
+      ?? .unknown
   }
 
   private func recordMemory(_ nodeID: UUID, _ entry: String) {
@@ -1172,10 +1239,10 @@ public actor GraphStore {
   /// telling every client the graph moved when nothing did.
   @discardableResult
   private func refreshPresence() async -> Bool {
-    guard let onReadPresence else { return false }
+    guard onReadPresence != nil else { return false }
     var changed = false
     for node in graph.nodes where !node.isResolved {
-      let reading = await onReadPresence(node, graph.project.path)
+      guard let reading = await presenceReading(of: node) else { continue }
       guard graph.nodes[id: node.id]?.presence != reading else { continue }
       graph.nodes[id: node.id]?.presence = reading
       changed = true
@@ -1831,7 +1898,18 @@ public actor GraphStore {
         announceError("mail watch refused: an empty topic is no topic — omit it")
         return
       }
-      graph.nodes[id: watcherID]?.mailroomWatch = MailroomWatch(topic: trimmed)
+      let watch = MailroomWatch(topic: trimmed)
+      graph.nodes[id: watcherID]?.mailroomWatch = watch
+      // Re-scoping is `--off` for the topic being left: a watch is one subscription, so
+      // the wakes staged under the old topic are for posts this loop is no longer
+      // asking about and arriving minutes later is the same defect turning the watch
+      // off had (issue #304). A wake that still matches the new scope stays queued, and
+      // a peer's `--follow-up` message is not a wake and is untouched.
+      pendingFollowUps.removeAll { pending in
+        guard pending.nodeID == watcherID, let postID = pending.watchedPostID else { return false }
+        guard let post = graph.mailroom.first(where: { $0.id == postID }) else { return true }
+        return !watch.matches(post.topic)
+      }
       recordMemory(
         watcherID, "mailroom: now watching \(trimmed.map { "'\($0)'" } ?? "all posts")")
     } else {
@@ -2151,8 +2229,7 @@ public actor GraphStore {
       return false
     }
     if RemoteProjectLocation.parse(projectPath: graph.project.path) != nil {
-      guard let onReadPresence else { return true }
-      let reading = await onReadPresence(node, graph.project.path)
+      guard let reading = await presenceReading(of: node) else { return true }
       if reading.presence == .absent { return true }
       recordMemory(
         nodeID,
@@ -2653,7 +2730,6 @@ public actor GraphStore {
     // lost to a daemon restart delays the message to the next wake instead of
     // dropping it.
     if followUp, deliversLater(to: target) {
-      recordMemory(nodeID, "follow-up staged: \(message)")
       pendingFollowUps.append(
         PendingFollowUp(nodeID: nodeID, text: message, watchedPostID: watchedPostID))
       return
@@ -2703,14 +2779,21 @@ public actor GraphStore {
   /// staged, exactly as before.
   static let respawnedSessionSettle: Duration = .seconds(3)
 
-  /// Whether a follow-up to this node should wait rather than type now. Mid-turn and
-  /// mid-check both qualify — deferring to a busy agent is the flag's entire meaning —
-  /// while an idle session gets ordinary immediate delivery and a dead one gets the
+  /// Whether a follow-up to this node joins the queue rather than being typed now.
+  /// Every live target does; only one the bus cannot reach at all falls through to the
   /// ordinary staged-to-memory path.
+  ///
+  /// This used to answer from the **cached** `node.presence` the poll last wrote, while
+  /// the drain took a live reading of its own — two readings of one question, and a
+  /// follow-up staged while they disagreed was typed in ahead of three already queued
+  /// (issue #311). Arrival order is the only promise a queue makes, so the drain's
+  /// reading is now the single decision point and this one has no presence in it.
+  /// Nothing waits longer for it: `handle` ends in a drain, so a follow-up to an idle
+  /// target is queued and handed over inside the same call — behind whatever was
+  /// already queued for that target, which is the whole difference.
   private func deliversLater(to target: LoopNode) -> Bool {
     switch MessageBus.deliverability(to: target) {
-    case .targetBusyWithACheck: return true
-    case nil: return target.presence?.presence != .idle
+    case .targetBusyWithACheck, nil: return true
     default: return false
     }
   }
@@ -2719,46 +2802,119 @@ public actor GraphStore {
   /// the store settles, and from the presence poll — the reading that says "idle" is
   /// the reading this waits for. A target that resolved or died is simply dropped from
   /// the queue: its memory log has carried the message since it was staged.
+  /// The memory record a deferred follow-up leaves, written the first time it is really
+  /// put back rather than when it was queued. Every live target's follow-up now joins
+  /// the queue and `handle` ends in a drain, so a message to an idle session is queued
+  /// and handed over inside the same call: a log line saying it was staged would
+  /// describe a wait that never happened, and a watcher's log would gain one per post it
+  /// was woken for. Anything that does wait is recorded before the pass that deferred it
+  /// ends. The floor under all of it is the board: a peer's message is mirrored onto
+  /// the Mailroom when it is sent and a watcher's wake is *about* a post already there,
+  /// so the content survives a daemon that dies mid-pass either way — this line is what
+  /// puts it in front of the loop's next wake without it having to go looking.
+  private func staged(_ pending: PendingFollowUp) -> PendingFollowUp {
+    guard !pending.recorded else { return pending }
+    recordMemory(pending.nodeID, "follow-up staged: \(pending.text)")
+    var recorded = pending
+    recorded.recorded = true
+    return recorded
+  }
+
   private func drainPendingFollowUps() async {
-    guard !pendingFollowUps.isEmpty, !isDrainingFollowUps else { return }
-    isDrainingFollowUps = true
-    defer { isDrainingFollowUps = false }
+    guard !pendingFollowUps.isEmpty else { return }
+    let taken = Date()
+    if let held = drainLease {
+      let heldFor = taken.timeIntervalSince(held)
+      guard heldFor >= drainLeaseDuration.timeInterval else { return }
+      // The line that stops frozen and working looking identical. A drain past its lease
+      // has hung somewhere this store could not bound — the delivery chain, most likely
+      // — and #289's diagnostics are where that becomes visible instead of being
+      // inferred from mail that never came.
+      DaemonLog.shared.record(
+        "drain-stall",
+        DaemonRequestContext.fields + [
+          ("held_ms", DaemonLog.milliseconds(heldFor)),
+          ("queued", String(pendingFollowUps.count)),
+        ])
+    }
+    drainLease = taken
+    // Released only if it is still ours: a successor that took over after this lease
+    // expired must not have its own lease cleared by the drain it replaced finally
+    // returning — the mistake `RemoteEnsureGate.end(_:token:)` documents.
+    defer { if drainLease == taken { drainLease = nil } }
     // Taken and cleared in one actor step, delivered from the local batch. Anything
     // queued while a delivery below is suspended lands in `pendingFollowUps` untouched
     // and is folded back in behind the retries at the end — never overwritten.
     let batch = pendingFollowUps
     pendingFollowUps = []
     var remaining: [PendingFollowUp] = []
-    for pending in batch {
-      guard let node = graph.nodes[id: pending.nodeID], !node.isResolved else { continue }
-      // A watcher's wake is only owed while the watch stands and the post is unread:
-      // `mail watch --off` after the wake was staged, or an inbox that has since read
-      // past the post, means the wake has nothing left to say.
+    var index = 0
+    // Folded back on the way out of every path, not only the one that runs to the end:
+    // the walk holds the whole queue in locals, and the stable-release check's reading
+    // of the wedge was that a drain which never returns strands them there. It cannot
+    // now — the reading below is bounded — but what the pass did not resolve belongs on
+    // the queue rather than in a variable about to go out of scope, whatever the exit.
+    defer { pendingFollowUps = remaining + Array(batch[index...]) + pendingFollowUps }
+    // One reading per target per pass, taken the first time this pass reaches that
+    // target and reused for the rest of its queue. Reading again between items is what
+    // let a turn ending *mid-drain* reorder the queue: three messages for one loop went
+    // out as [2, 3, 1] because the first was read while the session was still busy and
+    // the next two after it went idle. One drain produces that on its own, so #309's
+    // non-reentrancy guard cannot see it — and it is the [#357-before-#347] ordering
+    // issue #304 was filed for. A reading taken once cannot disagree with itself, so
+    // the batch either goes out in order or waits together for the next pass. It also
+    // costs one probe per target rather than one per message.
+    var readings: [UUID: Presence] = [:]
+    while index < batch.count {
+      let pending = batch[index]
+      index += 1
+      guard let node = graph.nodes[id: pending.nodeID], !node.isResolved else {
+        // Its work is over, but the message must not go with the queue entry: the log is
+        // what the loop's next wake reads, and for a resolved loop that is all there is.
+        if !pending.recorded, graph.nodes[id: pending.nodeID] != nil {
+          recordMemory(pending.nodeID, "while you were away: \(pending.text)")
+        }
+        continue
+      }
+      // A watcher's wake is only owed while the watch that asked for it stands and the
+      // post is unread: `mail watch --off` after the wake was staged, an inbox that has
+      // since read past the post, or a watch re-scoped to another topic — the wakes the
+      // abandoned topic staged are not this watcher's mail any more — all mean the wake
+      // has nothing left to say. A post the room has since evicted has nothing to point
+      // at either.
       if let postID = pending.watchedPostID {
-        guard node.mailroomWatch != nil, (node.lastMailroomRead ?? 0) < postID else { continue }
+        guard let watch = node.mailroomWatch, (node.lastMailroomRead ?? 0) < postID,
+          let post = graph.mailroom.first(where: { $0.id == postID }), watch.matches(post.topic)
+        else { continue }
       }
       switch MessageBus.deliverability(to: node) {
       case .targetBusyWithACheck:
-        remaining.append(pending)
+        remaining.append(staged(pending))
         continue
       case .some:
+        if !pending.recorded {
+          recordMemory(pending.nodeID, "while you were away: \(pending.text)")
+        }
         continue
       case nil:
         break
       }
-      let presence: Presence?
-      if let onReadPresence {
-        presence = await onReadPresence(node, graph.project.path).presence
+      let presence: Presence
+      if let known = readings[pending.nodeID] {
+        presence = known
       } else {
-        presence = node.presence?.presence
+        presence =
+          await presenceReading(of: node)?.presence ?? node.presence?.presence ?? .unknown
+        readings[pending.nodeID] = presence
       }
+      // `.idle` and nothing else, which is what makes a timed-out read safe: `.unknown`
+      // is not a state, so the message stays queued for a pass that gets an answer.
       guard presence == .idle else {
-        remaining.append(pending)
+        remaining.append(staged(pending))
         continue
       }
       _ = await deliverToSession(node, pending.text)
     }
-    pendingFollowUps = remaining + pendingFollowUps
   }
 
   private func announceError(_ message: String) {
@@ -3023,12 +3179,7 @@ public actor GraphStore {
       // the tree moves: re-delivering every poll would be a full agent turn a minute,
       // the unbounded spend the failure-tail dedup exists to prevent.
       if let fingerprint, goalCache.fingerprint(for: nodeID) == fingerprint {
-        let presence: Presence?
-        if let onReadPresence {
-          presence = await onReadPresence(node, graph.project.path).presence
-        } else {
-          presence = node.presence?.presence
-        }
+        let presence = await presenceReading(of: node)?.presence ?? node.presence?.presence
         // A nil presence stays skipped: the relay only ever tells a session it can see
         // idle, so falling through would pay the predicate's price for a wake that can
         // never land. Such a loop's exits are its stall bound and its human.
@@ -3117,12 +3268,7 @@ public actor GraphStore {
   ) async {
     let tail = outcome.outputTail.trimmingCharacters(in: .whitespacesAndNewlines)
     guard !tail.isEmpty, goalCache.feedback(for: node.id) != tail else { return }
-    let presence: Presence?
-    if let onReadPresence {
-      presence = await onReadPresence(node, graph.project.path).presence
-    } else {
-      presence = node.presence?.presence
-    }
+    let presence = await presenceReading(of: node)?.presence ?? node.presence?.presence
     guard presence == .idle, MessageBus.deliverability(to: node) == nil else { return }
     let message =
       "[graphcode] Goal not met yet: `\(predicate)` still exits non-zero. "
@@ -3220,12 +3366,7 @@ public actor GraphStore {
     guard node.backend.capabilities.supportsDaemonRecurrence || onHeartbeatEnabled?() == true
     else { return }
     guard MessageBus.deliverability(to: node) == nil else { return }
-    let presence: Presence?
-    if let onReadPresence {
-      presence = await onReadPresence(node, graph.project.path).presence
-    } else {
-      presence = node.presence?.presence
-    }
+    let presence = await presenceReading(of: node)?.presence ?? node.presence?.presence
     guard presence != .busy else { return }
     let task = node.heartbeatTask ?? ""
     _ = await deliverToSession(
diff --git a/graphcode/Tests/DrainWedgeTests.swift b/graphcode/Tests/DrainWedgeTests.swift
new file mode 100644
index 00000000..310af5cb
--- /dev/null
+++ b/graphcode/Tests/DrainWedgeTests.swift
@@ -0,0 +1,202 @@
+import ComposableArchitecture
+import Foundation
+import GraphcodeKit
+import Testing
+
+/// Issue #311's wedge, from the rig `StableSoakCheck` reproduced it with on a real
+/// `graphcoded` (`worktrees/stable-check-311`, `probe/311-drain-wedge`): a control run
+/// delivered a bystander's queued mail in ~15s, and the run that differs in **one
+/// closure** — node A's presence read hangs — never delivered it in 364 seconds, with a
+/// client attached throughout and zero errors logged.
+///
+/// Its tests asserted the freeze, since that was what beta5 did. These are the same
+/// scenarios with the assertions turned round to what this branch does instead, plus the
+/// one its report asked for that no deadline can give: a hang in a path with no deadline
+/// at all.
+@Suite
+struct DrainWedgeTests {
+  private struct Fixture {
+    let graph: LoopGraph
+    var hung: UUID { graph.nodes[0].id }
+    var bystander: UUID { graph.nodes[1].id }
+  }
+
+  private func fixture() -> Fixture {
+    Fixture(
+      graph: LoopGraph(
+        project: ProjectRef(path: "/tmp/drainwedge", name: "drainwedge"),
+        nodes: [
+          LoopNode(
+            title: "Hung", loopType: .goalBased, goal: GoalSpec(summary: "hangs"),
+            presence: PresenceReading(presence: .busy, confidence: .reported),
+            state: .running),
+          LoopNode(
+            title: "Bystander", loopType: .goalBased, goal: GoalSpec(summary: "innocent"),
+            presence: PresenceReading(presence: .busy, confidence: .reported),
+            state: .running),
+        ]))
+  }
+
+  private func settle(until flag: LockIsolated<Bool>) async {
+    for _ in 0..<2000 where !flag.value { try? await Task.sleep(for: .milliseconds(5)) }
+  }
+
+  /// The control, unchanged from the probe: node A's read answers `busy`, so the drain
+  /// steps past it and reaches node B, whose reading is idle.
+  @Test
+  func aBystandersFollowUpIsDeliveredWhenEveryPresenceReadAnswers() async {
+    let fixture = fixture()
+    let delivered = LockIsolated<[String]>([])
+    let store = GraphStore(
+      graph: fixture.graph,
+      onDeliverMessage: { _, message, _ in
+        delivered.withValue { $0.append(message) }
+        return true
+      },
+      onReadPresence: { node, _ in
+        PresenceReading(
+          presence: node.id == fixture.hung ? .busy : .idle, confidence: .reported)
+      })
+
+    await store.handle(
+      .messageNode(fixture.hung, text: "for the hung one", from: nil, followUp: true))
+    await store.handle(
+      .messageNode(fixture.bystander, text: "for the bystander", from: nil, followUp: true))
+
+    #expect(delivered.value == ["[graphcode] for the bystander"])
+  }
+
+  /// The wedge. Same graph, same two messages; node A's read never returns.
+  ///
+  /// On beta5 both the command that queued A's follow-up and every later drain were
+  /// held for the life of the process, and B's mail was never typed in however idle B
+  /// was. Bounded, the read is `.unknown` after the deadline, which is no state at all:
+  /// A's message stays owed and B's is delivered in the same pass.
+  @Test
+  func aHungPresenceReadDoesNotFreezeEveryOtherLoopsFollowUps() async {
+    let fixture = fixture()
+    let delivered = LockIsolated<[String]>([])
+    let announced = LockIsolated<[String]>([])
+    let entered = LockIsolated(false)
+    let release = LockIsolated(false)
+    let store = GraphStore(
+      graph: fixture.graph,
+      onDeliverMessage: { _, message, _ in
+        delivered.withValue { $0.append(message) }
+        return true
+      },
+      onReadPresence: { node, _ in
+        guard node.id == fixture.hung else {
+          return PresenceReading(presence: .idle, confidence: .reported)
+        }
+        entered.setValue(true)
+        while !release.value { try? await Task.sleep(for: .milliseconds(5)) }
+        return PresenceReading(presence: .busy, confidence: .reported)
+      },
+      onAnnounceError: { message in announced.withValue { $0.append(message) } },
+      presenceReadDeadline: .milliseconds(200))
+
+    // Queued in a task of its own so this test *fails* on the head where the drain never
+    // returns, rather than hanging with it.
+    let wedging = Task {
+      await store.handle(
+        .messageNode(fixture.hung, text: "for the hung one", from: nil, followUp: true))
+    }
+    await settle(until: entered)
+    await store.handle(
+      .messageNode(fixture.bystander, text: "for the bystander", from: nil, followUp: true))
+    // Past the deadline the wedged pass is over, and the next settle carries B's mail.
+    try? await Task.sleep(for: .milliseconds(400))
+    await store.handle(.memoNode(fixture.bystander, text: "settle", from: fixture.bystander))
+
+    #expect(delivered.value == ["[graphcode] for the bystander"])
+    #expect(announced.value.isEmpty)
+    release.setValue(true)
+    _ = await wedging.value
+  }
+
+  /// The same hang one layer up: `refreshPresence` walks the graph serially, so on beta5
+  /// the poll tick never reached the nodes behind the hung one and their presence stopped
+  /// being read at all. The bound is on the reading itself, so the walk continues.
+  @Test
+  func aHungPresenceReadDoesNotStopThePollTickReachingLaterLoops() async {
+    let fixture = fixture()
+    let asked = LockIsolated<[UUID]>([])
+    let entered = LockIsolated(false)
+    let release = LockIsolated(false)
+    let store = GraphStore(
+      graph: fixture.graph,
+      onReadPresence: { node, _ in
+        asked.withValue { $0.append(node.id) }
+        guard node.id == fixture.hung else {
+          return PresenceReading(presence: .idle, confidence: .reported)
+        }
+        entered.setValue(true)
+        while !release.value { try? await Task.sleep(for: .milliseconds(5)) }
+        return PresenceReading(presence: .busy, confidence: .reported)
+      },
+      presenceReadDeadline: .milliseconds(200))
+
+    let wedging = Task { await store.handle(.refreshUsage) }
+    await settle(until: entered)
+    try? await Task.sleep(for: .milliseconds(400))
+
+    #expect(asked.value.contains(fixture.hung))
+    #expect(asked.value.contains(fixture.bystander))
+    release.setValue(true)
+    _ = await wedging.value
+  }
+
+  /// What the deadline cannot reach, and the reason the guard is a lease.
+  ///
+  /// `deliverToSession` is the same `PTYProcessSession` chain as the presence read and
+  /// has no deadline of its own — `zmx ls`, a write per chunk, `ssh` for a remote loop.
+  /// A hang there holds a bare flag exactly as the presence read did, and the failure is
+  /// the one the soak measured: total, and silent. So the guard expires: the next drain
+  /// records the stall in the daemon log and takes the queue on.
+  @Test
+  func aHungDeliveryReleasesTheQueueWhenItsLeaseExpires() async {
+    let fixture = fixture()
+    let delivered = LockIsolated<[String]>([])
+    let lines = LockIsolated<[String]>([])
+    let entered = LockIsolated(false)
+    let release = LockIsolated(false)
+    let tap = DaemonLog.shared.tap { line in lines.withValue { $0.append(line) } }
+    defer { DaemonLog.shared.untap(tap) }
+    let store = GraphStore(
+      graph: fixture.graph,
+      onDeliverMessage: { node, message, _ in
+        guard node.id == fixture.hung else {
+          delivered.withValue { $0.append(message) }
+          return true
+        }
+        entered.setValue(true)
+        while !release.value { try? await Task.sleep(for: .milliseconds(5)) }
+        return true
+      },
+      onReadPresence: { _, _ in PresenceReading(presence: .idle, confidence: .reported) },
+      presenceReadDeadline: .milliseconds(200),
+      drainLeaseDuration: .milliseconds(300))
+
+    // The drain that hands A its message never returns from the delivery.
+    let wedging = Task {
+      await store.handle(
+        .messageNode(fixture.hung, text: "for the hung one", from: nil, followUp: true))
+    }
+    await settle(until: entered)
+
+    // While the lease stands, the queue is the wedged drain's: B waits.
+    await store.handle(
+      .messageNode(fixture.bystander, text: "for the bystander", from: nil, followUp: true))
+    #expect(delivered.value.isEmpty)
+
+    // Past the lease, the next drain takes over and says so.
+    try? await Task.sleep(for: .milliseconds(350))
+    await store.handle(.memoNode(fixture.bystander, text: "settle", from: fixture.bystander))
+
+    #expect(delivered.value == ["[graphcode] for the bystander"])
+    #expect(lines.value.contains { $0.contains("event=drain-stall") })
+    release.setValue(true)
+    _ = await wedging.value
+  }
+}
diff --git a/graphcode/Tests/FollowUpQueueOrderTests.swift b/graphcode/Tests/FollowUpQueueOrderTests.swift
new file mode 100644
index 00000000..dbedfc15
--- /dev/null
+++ b/graphcode/Tests/FollowUpQueueOrderTests.swift
@@ -0,0 +1,300 @@
+import ComposableArchitecture
+import Foundation
+import GraphcodeKit
+import MailroomKit
+import Testing
+
+#if canImport(Darwin)
+  import Darwin
+#endif
+
+/// The seam #309 left behind: how a staged follow-up decides *when* to be delivered.
+///
+/// One test per defect, each failing on the head before this change for its own reason —
+/// a combined test that passes says nothing about which half works:
+///
+/// - `aHungPresenceReadDoesNotStopDeliveryToOtherLoops` — #311, the unbounded read.
+/// - `tenStagedAndSixLaterMessagesArriveInSendOrder` — #311, the two readings.
+/// - `aTurnEndingMidDrainDoesNotReorderTheQueue` — #304 residual, ordering.
+/// - `reScopingAWatchDropsTheAbandonedTopicsStagedWakes` — #304 residual, re-scoping.
+/// - `aBacklogDrainsInOnePassRatherThanOnePerTick` — the rate the field measurement found.
+@Suite
+struct FollowUpQueueOrderTests {
+  /// What each target's session appears to be doing. Every knob is set by the test
+  /// before the pass it applies to, so nothing here depends on how many times the store
+  /// happens to ask — which is the difference between the two heads this must separate.
+  private actor Readings {
+    private var fallback: Presence = .busy
+    private var first: Presence?
+    private var reads = 0
+    private var delay: Duration = .zero
+    private var hang: Duration = .zero
+    private var hung: UUID?
+
+    func set(_ presence: Presence, delay: Duration = .zero) {
+      fallback = presence
+      first = nil
+      reads = 0
+      self.delay = delay
+    }
+
+    /// The turn boundary: the pass's first reading says `first`, everything after it
+    /// says `rest`. A drain that reads once per target per pass sees only `first`; one
+    /// that reads per item sees the session change under it mid-batch.
+    func firstReadThen(_ first: Presence, _ rest: Presence) {
+      self.first = first
+      fallback = rest
+      reads = 0
+    }
+
+    func hang(_ node: UUID, for duration: Duration) {
+      hung = node
+      hang = duration
+    }
+    func stopHanging() { hung = nil }
+
+    func read(_ node: UUID) async -> PresenceReading {
+      if node == hung { try? await Task.sleep(for: hang) }
+      if delay > .zero { try? await Task.sleep(for: delay) }
+      reads += 1
+      let presence = reads == 1 ? (first ?? fallback) : fallback
+      return PresenceReading(presence: presence, confidence: .reported)
+    }
+  }
+
+  /// A connection with a channel lifecycle of its own and a reader draining it, so the
+  /// presence poll runs — the rig `FollowUpDrainTests` needs, for the same reason.
+  private final class Attachment: @unchecked Sendable {
+    let daemonEnd: Int32
+    private let peer: Int32
+    private let drainer: Task<Void, Never>
+    init() {
+      var pair: [Int32] = [0, 0]
+      _ = socketpair(AF_UNIX, SOCK_STREAM, 0, &pair)
+      daemonEnd = pair[0]
+      peer = pair[1]
+      let reading = pair[1]
+      drainer = Task.detached {
+        var sink = [UInt8](repeating: 0, count: 65536)
+        while recv(reading, &sink, sink.count, 0) > 0 {}
+      }
+    }
+    deinit {
+      OutboundChannels.close(daemonEnd)
+      close(peer)
+      drainer.cancel()
+    }
+  }
+
+  private func makeStore(
+    readings: Readings, delivered: LockIsolated<[String]>, attachment: Attachment,
+    titles: [String] = ["Target", "Sender"], deadline: Duration = .seconds(45)
+  ) async -> (GraphStore, ids: [UUID]) {
+    let store = GraphStore(
+      onEnsureSession: { _, _ in },
+      onDeliverMessage: { _, message, _ in
+        delivered.withValue { $0.append(message) }
+        return true
+      },
+      onReadPresence: { node, _ in await readings.read(node.id) },
+      onMailroomEnabled: { true },
+      presenceReadDeadline: deadline)
+    for title in titles {
+      await store.handle(
+        .createNode(NodeDraft(title: title, loopType: .turnBased, firstInstruction: "Work")))
+    }
+    await store.addConnection(id: UUID(), fileDescriptor: attachment.daemonEnd)
+    // One poll so every target carries a `busy` reading: that is what stages a follow-up
+    // rather than typing it in, on this head and on the one before it.
+    await store.pollPresence()
+    return (store, await store.graph.nodes.map(\.id))
+  }
+
+  /// Settles the store — every command ends in the drain — without refreshing presence
+  /// on the way, so the drain's own readings are the only ones the test's script sees.
+  /// `.refreshUsage` polls presence first, which is right for the poll and wrong here.
+  private func settle(_ store: GraphStore, _ nodeID: UUID) async {
+    await store.handle(.memoNode(nodeID, text: "settle", from: nodeID))
+  }
+
+  private func plain(_ delivered: [String]) -> [String] {
+    delivered.map { $0.replacingOccurrences(of: "[graphcode] Sender: ", with: "") }
+  }
+
+  /// Issue #311, the presence read with no deadline — the blocker, because a wedged
+  /// drain and an empty one report exactly the same thing from outside.
+  ///
+  /// `onReadPresence` reaches `PTYProcessSession.waitCollectingOutput`, which ends only
+  /// when the probe's `terminationHandler` closes the stream; `ssh`'s `ConnectTimeout`
+  /// bounds the connect, not a command hanging on a host that has gone away. One such
+  /// read held `isDrainingFollowUps` for the life of the daemon, so mail for **every
+  /// other loop in the project** stopped — which is what this measures: not that the
+  /// wedged loop recovers, but that its neighbour is not taken down with it.
+  ///
+  /// Fails before the fix on elapsed time: the pass waits out the hung read.
+  @Test
+  func aHungPresenceReadDoesNotStopDeliveryToOtherLoops() async {
+    let readings = Readings()
+    let delivered = LockIsolated<[String]>([])
+    let attachment = Attachment()
+    let (store, ids) = await makeStore(
+      readings: readings, delivered: delivered, attachment: attachment,
+      titles: ["Wedged", "Neighbour", "Sender"], deadline: .milliseconds(200))
+    let (wedged, neighbour, sender) = (ids[0], ids[1], ids[2])
+
+    await store.handle(
+      .messageNode(wedged, text: "for the wedged loop", from: sender, followUp: true))
+    await store.handle(
+      .messageNode(neighbour, text: "for the neighbour", from: sender, followUp: true))
+    #expect(delivered.value.isEmpty)
+
+    // The wedged loop's session stops answering; every other read is instant and idle.
+    await readings.hang(wedged, for: .seconds(5))
+    await readings.set(.idle)
+    let started = Date()
+    await store.handle(.refreshUsage)
+    let elapsed = Date().timeIntervalSince(started)
+
+    #expect(plain(delivered.value) == ["for the neighbour"])
+    #expect(elapsed < 2)
+    // Not delivered blindly either: a read that ran out of time is `.unknown`, which is
+    // no state at all, so the message is still owed rather than spent.
+    await readings.stopHanging()
+    await store.handle(.refreshUsage)
+    #expect(plain(delivered.value) == ["for the neighbour", "for the wedged loop"])
+  }
+
+  /// Issue #311, the two readings — and the shape `MailDeliveryCheck` measured against
+  /// the live 0.1.64-beta5 daemon: ten messages sat staged and drained 77 to 147 seconds
+  /// later while six sent afterwards arrived within half a second, because those six
+  /// took the immediate path off a cached `idle` while the queue was still being worked
+  /// through. Order was perfect within each path and meaningless across them.
+  ///
+  /// Reproduced the same way here: the poll writes `idle` into the cache, and the later
+  /// messages arrive while that poll's own drain is still running.
+  ///
+  /// Fails before the fix on order — the six overtake the ten.
+  @Test
+  func tenStagedAndSixLaterMessagesArriveInSendOrder() async {
+    let readings = Readings()
+    let delivered = LockIsolated<[String]>([])
+    let attachment = Attachment()
+    let (store, ids) = await makeStore(
+      readings: readings, delivered: delivered, attachment: attachment)
+    let (target, sender) = (ids[0], ids[1])
+    for index in 1...10 {
+      await store.handle(
+        .messageNode(target, text: "staged \(index)", from: sender, followUp: true))
+    }
+    #expect(delivered.value.isEmpty)
+
+    // Idle now, and slow to answer: the poll's refresh writes `idle` into the cache
+    // inside the first 400ms, and its drain is still going at 600ms.
+    await readings.set(.idle, delay: .milliseconds(200))
+    async let poll: Void = store.pollPresence()
+    try? await Task.sleep(for: .milliseconds(600))
+    for index in 1...6 {
+      await store.handle(
+        .messageNode(target, text: "later \(index)", from: sender, followUp: true))
+    }
+    await poll
+    await store.pollPresence()
+
+    let expected =
+      (1...10).map { "staged \($0)" } + (1...6).map { "later \($0)" }
+    #expect(plain(delivered.value) == expected)
+  }
+
+  /// Issue #304's residual: a single drain reorders when the target's turn ends while
+  /// the drain is running. The first item is read mid-turn and put back; the next two are
+  /// read after the turn ends and go out — [2, 3, 1], with no overlapping drain anywhere,
+  /// which is why #309's non-reentrancy guard cannot see it.
+  ///
+  /// The reading is now taken once per target per pass, so a batch either goes out in
+  /// order or waits together for a pass that can deliver all of it.
+  ///
+  /// Fails before the fix on order: `["staged 2", "staged 3", "staged 1"]`.
+  @Test
+  func aTurnEndingMidDrainDoesNotReorderTheQueue() async {
+    let readings = Readings()
+    let delivered = LockIsolated<[String]>([])
+    let attachment = Attachment()
+    let (store, ids) = await makeStore(
+      readings: readings, delivered: delivered, attachment: attachment)
+    let (target, sender) = (ids[0], ids[1])
+    for index in 1...3 {
+      await store.handle(
+        .messageNode(target, text: "staged \(index)", from: sender, followUp: true))
+    }
+    #expect(delivered.value.isEmpty)
+
+    // The turn ends between the first item and the second.
+    await readings.firstReadThen(.busy, .idle)
+    await settle(store, target)
+
+    await readings.set(.idle)
+    await settle(store, target)
+
+    #expect(plain(delivered.value) == ["staged 1", "staged 2", "staged 3"])
+  }
+
+  /// Issue #304's other residual: `mail watch --on --topic other` re-scopes an existing
+  /// watch, and the wakes staged under the topic it left kept arriving — the same defect
+  /// `--off` had, reached by the command that does not read like a teardown.
+  ///
+  /// Fails before the fix by delivering the abandoned topic's wake as well.
+  @Test
+  func reScopingAWatchDropsTheAbandonedTopicsStagedWakes() async {
+    let readings = Readings()
+    let delivered = LockIsolated<[String]>([])
+    let attachment = Attachment()
+    let (store, ids) = await makeStore(
+      readings: readings, delivered: delivered, attachment: attachment)
+    let (target, sender) = (ids[0], ids[1])
+
+    await store.handle(.mailroomWatch(on: true, topic: "alpha", from: target))
+    await store.handle(.mailroomPost(text: "an alpha finding", topic: "alpha", from: sender))
+    // Re-scoped, without ever passing through `--off`.
+    await store.handle(.mailroomWatch(on: true, topic: "beta", from: target))
+    await store.handle(.mailroomPost(text: "a beta finding", topic: "beta", from: sender))
+    #expect(delivered.value.isEmpty)
+
+    await readings.set(.idle)
+    await store.pollPresence()
+
+    #expect(delivered.value.count == 1)
+    #expect(delivered.value.first?.contains("a beta finding") == true)
+    #expect(!delivered.value.contains { $0.contains("an alpha finding") })
+  }
+
+  /// The rate `MailDeliveryCheck` measured on the live daemon and nobody had filed:
+  /// about **two deliveries per poll tick**, so ten staged messages took 147 seconds.
+  /// Delivering into a session is what makes it busy, and a drain that re-read presence
+  /// between items therefore stopped after the first delivery of every pass and waited
+  /// for the next tick — the queue drained at one item per fifteen seconds.
+  ///
+  /// One reading per target per pass ends it: the pass judges the target once, before it
+  /// has typed anything into it, and hands over the whole backlog in that pass.
+  ///
+  /// Fails before the fix on count: one delivered per pass, not ten.
+  @Test
+  func aBacklogDrainsInOnePassRatherThanOnePerTick() async {
+    let readings = Readings()
+    let delivered = LockIsolated<[String]>([])
+    let attachment = Attachment()
+    let (store, ids) = await makeStore(
+      readings: readings, delivered: delivered, attachment: attachment)
+    let (target, sender) = (ids[0], ids[1])
+    for index in 1...10 {
+      await store.handle(
+        .messageNode(target, text: "backlog \(index)", from: sender, followUp: true))
+    }
+    #expect(delivered.value.isEmpty)
+
+    // Idle when the pass begins, busy from the moment the first message is typed in.
+    await readings.firstReadThen(.idle, .busy)
+    await settle(store, target)
+
+    #expect(plain(delivered.value) == (1...10).map { "backlog \($0)" })
+  }
+}
diff --git a/graphcode/Tests/OrderingProbeTests.swift b/graphcode/Tests/OrderingProbeTests.swift
new file mode 100644
index 00000000..0eb4412e
--- /dev/null
+++ b/graphcode/Tests/OrderingProbeTests.swift
@@ -0,0 +1,225 @@
+import ComposableArchitecture
+import Foundation
+import GraphcodeKit
+import MailroomKit
+import Testing
+
+#if canImport(Darwin)
+  import Darwin
+#endif
+
+/// Issue #304, ordering half. #309 fixed duplicates and loss; these three probes
+/// characterise what it did not fix. Each is deterministic — the presence readings are
+/// scripted, so a failure here is a defect, not a race that happened to land.
+@Suite
+struct OrderingProbeTests {
+  /// Answers each presence read from a script, so a turn can end at an exact point in
+  /// one drain's walk over its batch.
+  private actor ScriptedReadings {
+    private var script: [Presence]
+    private var index = 0
+    private(set) var log: [String] = []
+    init(_ script: [Presence]) { self.script = script }
+    func read(_ scripted: Bool) -> PresenceReading {
+      guard scripted else { return PresenceReading(presence: .idle, confidence: .reported) }
+      let presence = index < script.count ? script[index] : (script.last ?? .idle)
+      index += 1
+      log.append("\(index): \(presence)")
+      return PresenceReading(presence: presence, confidence: .reported)
+    }
+    func trace() -> [String] { log }
+  }
+
+  private final class Attachment: @unchecked Sendable {
+    let daemonEnd: Int32
+    private let peer: Int32
+    private let drainer: Task<Void, Never>
+    init() {
+      var pair: [Int32] = [0, 0]
+      _ = socketpair(AF_UNIX, SOCK_STREAM, 0, &pair)
+      daemonEnd = pair[0]
+      peer = pair[1]
+      let reading = pair[1]
+      drainer = Task.detached {
+        var sink = [UInt8](repeating: 0, count: 65536)
+        while recv(reading, &sink, sink.count, 0) > 0 {}
+      }
+    }
+    deinit {
+      OutboundChannels.close(daemonEnd)
+      close(peer)
+      drainer.cancel()
+    }
+  }
+
+  private func makeStore(
+    readings: ScriptedReadings, delivered: LockIsolated<[String]>, attachment: Attachment
+  ) async -> (GraphStore, target: UUID, sender: UUID) {
+    let store = GraphStore(
+      onEnsureSession: { _, _ in },
+      onDeliverMessage: { _, message, _ in
+        delivered.withValue { $0.append(message) }
+        return true
+      },
+      onReadPresence: { node, _ in await readings.read(node.title == "Target") },
+      onMailroomEnabled: { true })
+    await store.handle(
+      .createNode(NodeDraft(title: "Target", loopType: .turnBased, firstInstruction: "Work")))
+    await store.handle(
+      .createNode(NodeDraft(title: "Sender", loopType: .turnBased, firstInstruction: "Work")))
+    let ids = await store.graph.nodes.map(\.id)
+    await store.addConnection(id: UUID(), fileDescriptor: attachment.daemonEnd)
+    return (store, ids[0], ids[1])
+  }
+
+  private func plain(_ delivered: LockIsolated<[String]>) -> [String] {
+    delivered.value.map { $0.replacingOccurrences(of: "[graphcode] Sender: ", with: "") }
+  }
+
+  /// PROBE A — a turn ending mid-drain reorders, with a SINGLE drain.
+  ///
+  /// The first read of the walk says busy, so item 1 is put back; the reads for items 2
+  /// and 3 say idle, so they are delivered. Item 1 then arrives on the next drain. No
+  /// two drains overlap, so #309's in-flight guard is not involved at all.
+  @Test
+  func aTurnEndingMidDrainReordersWithinOneDrain() async {
+    // Target reads, in order: poll-1 refresh (busy, so every message stages), then the
+    // settle drain that follows each `messageNode`. The third of those drains walks a
+    // batch of two and the turn ends between its two reads: busy for item 1, idle for
+    // item 2. One drain, no overlap — #309's in-flight guard cannot see this.
+    let readings = ScriptedReadings([.busy, .busy, .busy, .idle, .idle])
+    let delivered = LockIsolated<[String]>([])
+    let attachment = Attachment()
+    let (store, target, sender) = await makeStore(
+      readings: readings, delivered: delivered, attachment: attachment)
+    await store.pollPresence()
+    for index in 1...3 {
+      await store.handle(
+        .messageNode(target, text: "staged \(index)", from: sender, followUp: true))
+    }
+    await store.pollPresence()
+
+    let texts = plain(delivered)
+    print("PROBE-A order: \(texts) reads: \(await readings.trace())")
+    #expect(texts.count == 3)
+    #expect(texts == ["staged 1", "staged 2", "staged 3"])
+  }
+
+  /// PROBE B — a later follow-up jumps the queue.
+  ///
+  /// `deliversLater` reads the CACHED `node.presence`; the drain reads LIVE per item.
+  /// Cache says busy so message 1 stages; the live read in the drain also says busy so
+  /// it stays queued; then message 2 arrives while the cache still says busy — it too
+  /// stages, so far so good. The jump is the other disagreement: the cache goes idle
+  /// (the poll wrote it) while the live read is still busy, so message 2 is typed
+  /// straight in ahead of the still-queued message 1.
+  @Test
+  func aLaterFollowUpDoesNotJumpAnEarlierOne() async {
+    // Target reads: poll-1 refresh busy (so "first" stages), its settle drain busy (so
+    // "first" is requeued), poll-2 refresh idle — the CACHE is now idle — and poll-2's
+    // drain read busy, so "first" stays queued. "second" then consults the cache, sees
+    // idle, and is typed straight into the session ahead of the still-queued "first".
+    let readings = ScriptedReadings([.busy, .busy, .idle, .busy, .idle])
+    let delivered = LockIsolated<[String]>([])
+    let attachment = Attachment()
+    let (store, target, sender) = await makeStore(
+      readings: readings, delivered: delivered, attachment: attachment)
+    await store.pollPresence()
+    await store.handle(.messageNode(target, text: "first", from: sender, followUp: true))
+
+    await store.pollPresence()
+    await store.handle(.messageNode(target, text: "second", from: sender, followUp: true))
+    await store.pollPresence()
+
+    let texts = plain(delivered)
+    print("PROBE-B order: \(texts) reads: \(await readings.trace())")
+    #expect(texts == ["first", "second"])
+  }
+
+  /// PROBE C — re-scoping a watch keeps the abandoned topic's staged wakes.
+  ///
+  /// `mail watch --on --topic other` overwrites the subscription but never touches the
+  /// wakes already queued for the old scope, and the drain's guard only asks whether
+  /// *some* watch stands, never whether it still matches the post's topic.
+  @Test
+  func reScopingAWatchDropsTheAbandonedTopicsWakes() async {
+    // Target reads: poll-1 refresh busy (so the wake stages) and the settle drain that
+    // follows the post busy too, so the wake is still queued when the re-scope lands.
+    let readings = ScriptedReadings([.busy, .busy, .idle, .idle])
+    let delivered = LockIsolated<[String]>([])
+    let attachment = Attachment()
+    let (store, target, sender) = await makeStore(
+      readings: readings, delivered: delivered, attachment: attachment)
+    await store.pollPresence()
+    await store.handle(.mailroomWatch(on: true, topic: "alpha", from: target))
+    await store.handle(.mailroomPost(text: "an alpha post", topic: "alpha", from: sender))
+    #expect(delivered.value.isEmpty)
+
+    // The watcher re-scopes to a different topic before the wake could land.
+    await store.handle(.mailroomWatch(on: true, topic: "beta", from: target))
+    await store.pollPresence()
+
+    let texts = plain(delivered)
+    print("PROBE-C delivered: \(texts) reads: \(await readings.trace())")
+    #expect(texts.isEmpty)
+    #expect(texts.isEmpty)
+  }
+
+  /// PROBE D — the queue jump needs no exotic timing at all.
+  ///
+  /// Delivering an item is *itself* what makes the target busy, so a drain that hands
+  /// over item 1 will read busy for item 2 and requeue it. The cache, meanwhile, was
+  /// written `idle` by the refresh that opened the same poll and is not rewritten until
+  /// the next one — up to fifteen seconds later. Every follow-up that arrives in that
+  /// window consults the stale idle cache and is typed straight in, ahead of the item
+  /// still queued. This is the ordinary shape of a partly-drained queue, not a race.
+  @Test
+  func aFollowUpArrivingAfterAPartialDrainDoesNotJumpTheRemainder() async {
+    // Target reads: poll-1 refresh busy (both messages stage) and its two settle-drain
+    // reads busy; poll-2 refresh idle (cache goes idle), poll-2 drain: item 1 idle —
+    // delivered, which starts a turn — item 2 busy, requeued.
+    let readings = ScriptedReadings([.busy, .busy, .busy, .idle, .idle, .busy, .idle, .idle])
+    let delivered = LockIsolated<[String]>([])
+    let attachment = Attachment()
+    let (store, target, sender) = await makeStore(
+      readings: readings, delivered: delivered, attachment: attachment)
+    await store.pollPresence()
+    for index in 1...2 {
+      await store.handle(
+        .messageNode(target, text: "staged \(index)", from: sender, followUp: true))
+    }
+    await store.pollPresence()
+    // A peer speaks while the target is mid-turn on what the drain just handed it.
+    await store.handle(.messageNode(target, text: "third", from: sender, followUp: true))
+    await store.pollPresence()
+
+    let texts = plain(delivered)
+    print("PROBE-D order: \(texts) reads: \(await readings.trace())")
+    #expect(texts == ["staged 1", "staged 2", "third"])
+  }
+
+  /// PROBE E — the workaround for probe C: `--off` then `--on --topic beta` is clean.
+  ///
+  /// `--off` is the only path that reaches the staged wakes, so a re-scope spelled as
+  /// two commands drops the abandoned topic's backlog where the one-command re-scope
+  /// keeps it. Worth knowing, because it is what an operator can do today.
+  @Test
+  func offThenOnIsACleanReScope() async {
+    let readings = ScriptedReadings([.busy, .busy, .idle, .idle])
+    let delivered = LockIsolated<[String]>([])
+    let attachment = Attachment()
+    let (store, target, sender) = await makeStore(
+      readings: readings, delivered: delivered, attachment: attachment)
+    await store.pollPresence()
+    await store.handle(.mailroomWatch(on: true, topic: "alpha", from: target))
+    await store.handle(.mailroomPost(text: "an alpha post", topic: "alpha", from: sender))
+
+    await store.handle(.mailroomWatch(on: false, topic: nil, from: target))
+    await store.handle(.mailroomWatch(on: true, topic: "beta", from: target))
+    await store.pollPresence()
+
+    let texts = plain(delivered)
+    print("PROBE-E delivered: \(texts) reads: \(await readings.trace())")
+    #expect(texts.isEmpty)
+  }
+}