fix(buzz-acp): bound busy-owner hold to prevent cross-channel starvation (#7337)

## Problem

#6732 added a busy-owner hold to the ACP harness: when a scope's
recorded session owner (`session_owners`) is checked out on **any**
turn, `dispatch_pending` holds the scope's batch instead of dispatching
it. The hold was added to keep one provider session per thread — but it
is unconditional: it applies to `Conversation` scopes too, and it has no
time bound.

Under the default `session_policy=channel`, every channel collapses to a
single `Conversation` scope, so once two channels' sessions land on the
same worker (pass 2 of `try_claim` picks the first idle worker by index,
so this happens quickly after any restart), channel A's mention starves
behind channel B's in-flight turn — for up to the full
`max_turn_duration` (7200s by default) — while other workers sit idle.
The only signal is a DEBUG-level log, and the 👀 seen-reaction is added
at queue admission *before* the hold decision, so the user sees the
agent acknowledge the mention and then nothing.

Observed in production on the first day of the v0.5.22 rollout: three
separate incidents where a mention got 👀 but no turn started until an
unrelated channel's turn ended on the shared worker (in the worst case
the blocking turn sat in a single tool call for 6+ minutes).

## Fix

One new seam, `AgentPool::hold_decision`, replaces the raw
`should_hold_for_busy_owner` check in `dispatch_pending` (the predicate
itself is unchanged and remains the inner check):

- **`Conversation` scopes never hold.** Channel-policy channels and all
DMs dispatch immediately; a busy owner means forking onto an idle
worker, exactly the pre-#6732 behavior. This removes the cross-channel
head-of-line blocking entirely for the default policy.
- **`Thread` scopes hold for a bounded window.**
`HOLD_BUSY_OWNER_TIMEOUT` (10s) is measured from the first time the
batch is held (`held_since` stamp); once elapsed, the batch stops
holding and forks a fresh session on an idle worker, rebuilding thread
context from the relay. This preserves #6732's session-continuity intent
for the momentary-busy case while capping the worst-case wait. No new
timer is needed: held batches are requeued with preserved timestamps and
re-evaluated on every dispatch trigger (turn end, relay event, 30s
maintenance tick), so the effective worst-case re-check gap on a fully
silent system is one maintenance tick.
- **Holds are observable.** Holding logs at INFO and a hold expiry logs
at WARN (previously DEBUG-only), and both emit observer-feed events
(`busy_owner_hold`, `busy_owner_hold_forked`) with the scope, owner
index, and held duration.

`held_since` is derived state and is cleared on every removal path:
dispatch/fork (inside `hold_decision`), `invalidate_channel_sessions`,
`invalidate_scope_session`, and `switch_idle_agent_model`.

## Accepted trade-offs

- A fork after an expired hold leaves the old owner's now-orphaned
thread session in its session map until natural rotation/invalidation —
benign, and identical to pre-#6732 fork semantics (`loadSession: false`;
sessions are worker-pinned, so migration is not an option).
- Under sustained pool exhaustion the hold stamp is cleared on the fork
attempt and re-stamped next cycle, so the bound is effectively "timeout
after a worker frees up," not absolute wall clock.

## Tests

- New table test `hold_decision_covers_variant_session_busy_and_timeout`
over the full input space (scope variant × idle-session presence × owner
busyness × elapsed vs. window). The `Conversation` + busy-owner row is
the cross-channel regression guard; the past-window row guards the
bound. Both were mutation-checked: removing the variant gate or the
timeout branch fails the suite.
- `busy_session_owner_holds_batch_instead_of_forking_session` extended
with the Hold → ForkAfterHold transition, the `Conversation` dispatch
guard, and `held_since` pruning on channel invalidation.
- Scope-invalidation and idle-model-switch tests extended to cover
`held_since` cleanup alongside the existing `session_owners` assertions.

Signed-off-by: Will Pfleger <pfleger.will@gmail.com>

crates/buzz-acp/src/lib.rs

@@ -3049,9 +3049,13 @@ async fn tokio_main() -> Result<()> {
             // called on relay events or pool results, neither of which
             // arrive when the channel is silent.
             if queue.has_flushable_work() {
-                for (scope, thread_tags) in
-                    dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity)
-                {
+                for (scope, thread_tags) in dispatch_pending(
+                    &mut pool,
+                    &mut queue,
+                    &ctx,
+                    &mut last_activity,
+                    observer.as_ref(),
+                ) {
                     typing_channels.insert(scope, thread_tags);
                 }
             }
@@ -3101,9 +3105,13 @@ async fn tokio_main() -> Result<()> {
         // this, batches requeued during crash recovery sit idle until the
         // next relay event arrives — which can be minutes on quiet channels.
         if respawn_collected {
-            for (scope, thread_tags) in
-                dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity)
-            {
+            for (scope, thread_tags) in dispatch_pending(
+                &mut pool,
+                &mut queue,
+                &ctx,
+                &mut last_activity,
+                observer.as_ref(),
+            ) {
                 typing_channels.insert(scope, thread_tags);
             }
         }
@@ -3541,7 +3549,7 @@ async fn tokio_main() -> Result<()> {
                             );
                             if pool_ready {
                                 for (scope, thread_tags) in
-                                    dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity)
+                                    dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, observer.as_ref())
                                 {
                                     typing_channels.insert(scope, thread_tags);
                                 }
@@ -3641,7 +3649,7 @@ async fn tokio_main() -> Result<()> {
                     } else if queue.has_flushable_work() {
                         tracing::debug!("heartbeat_skipped_events");
                         for (scope, thread_tags) in
-                            dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity)
+                            dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, observer.as_ref())
                         {
                             typing_channels.insert(scope, thread_tags);
                         }
@@ -3742,9 +3750,13 @@ async fn tokio_main() -> Result<()> {
                 {
                     break;
                 }
-                for (scope, thread_tags) in
-                    dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity)
-                {
+                for (scope, thread_tags) in dispatch_pending(
+                    &mut pool,
+                    &mut queue,
+                    &ctx,
+                    &mut last_activity,
+                    observer.as_ref(),
+                ) {
                     typing_channels.insert(scope, thread_tags);
                 }
             }
@@ -3767,9 +3779,13 @@ async fn tokio_main() -> Result<()> {
                     tracing::error!("all agents dead — exiting");
                     break;
                 }
-                for (scope, thread_tags) in
-                    dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity)
-                {
+                for (scope, thread_tags) in dispatch_pending(
+                    &mut pool,
+                    &mut queue,
+                    &ctx,
+                    &mut last_activity,
+                    observer.as_ref(),
+                ) {
                     typing_channels.insert(scope, thread_tags);
                 }
             }
@@ -3921,9 +3937,13 @@ async fn tokio_main() -> Result<()> {
                 // tear down the in-flight task; on its completion the
                 // queue drains. We still try here in case the in-flight
                 // task has already returned.
-                for (scope, thread_tags) in
-                    dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity)
-                {
+                for (scope, thread_tags) in dispatch_pending(
+                    &mut pool,
+                    &mut queue,
+                    &ctx,
+                    &mut last_activity,
+                    observer.as_ref(),
+                ) {
                     typing_channels.insert(scope, thread_tags);
                 }
             }
@@ -3949,9 +3969,13 @@ async fn tokio_main() -> Result<()> {
                             "ready",
                             None,
                         );
-                        for (scope, thread_tags) in
-                            dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity)
-                        {
+                        for (scope, thread_tags) in dispatch_pending(
+                            &mut pool,
+                            &mut queue,
+                            &ctx,
+                            &mut last_activity,
+                            observer.as_ref(),
+                        ) {
                             typing_channels.insert(scope, thread_tags);
                         }
                     }
@@ -4340,6 +4364,7 @@ fn dispatch_pending(
     queue: &mut EventQueue,
     ctx: &Arc<PromptContext>,
     last_activity: &mut tokio::time::Instant,
+    observer: Option<&observer::ObserverHandle>,
 ) -> Vec<(scope::SessionScope, ThreadTags)> {
     // Keyed by the exact session scope, not the channel: two threads dispatching
     // concurrently in one channel get distinct typing entries so completing one
@@ -4348,9 +4373,13 @@ fn dispatch_pending(
     // Batches held back this cycle because the worker that owns their thread's
     // session is busy. They stay flushed-out of the queue (in-flight) until we
     // release them at the end so `flush_next` cannot re-pick them mid-loop;
-    // releasing requeues them so the next dispatch (when the owner returns)
-    // reuses that exact session instead of forking a duplicate.
+    // releasing requeues them so the next dispatch (when the owner returns, or
+    // once the bounded hold expires) reuses that exact session or forks a fresh
+    // one instead of starving.
     let mut held: Vec<FlushBatch> = Vec::new();
+    // One clock read for the whole cycle so every batch's bounded-hold window is
+    // measured against the same instant.
+    let now = std::time::Instant::now();
     loop {
         let batch = match queue.flush_next() {
             Some(b) => b,
@@ -4358,17 +4387,68 @@ fn dispatch_pending(
         };
         let channel_id = batch.channel_id;
         let scope = batch.scope.clone();
-        // Authoritative affinity: if the worker that owns this thread's session
-        // is checked out (busy on another turn), hold the batch rather than let
-        // an idle worker open a second session for the same thread.
-        if pool.should_hold_for_busy_owner(&scope) {
-            tracing::debug!(
-                channel = %channel_id,
-                scope = %scope.telemetry_label(),
-                "holding batch — session owner busy; awaiting its return to avoid duplicate session"
-            );
-            held.push(batch);
-            continue;
+        // Authoritative affinity, variant-gated and bounded: only a `Thread`
+        // scope whose session owner is checked out (busy on another turn) is
+        // held, and only until `HOLD_BUSY_OWNER_TIMEOUT` elapses. `Conversation`
+        // scopes never hold — a busy owner there forks onto another idle worker,
+        // so an active channel cannot starve a sibling channel on a shared
+        // worker. A held thread that outwaits the window forks a fresh session
+        // rather than starve behind an unbounded turn.
+        match pool.hold_decision(&scope, now, pool::HOLD_BUSY_OWNER_TIMEOUT) {
+            pool::HoldDecision::Hold {
+                held_for,
+                owner_index,
+            } => {
+                tracing::info!(
+                    channel = %channel_id,
+                    scope = %scope.telemetry_label(),
+                    owner_index,
+                    held_for_secs = held_for.as_secs_f64(),
+                    "busy-owner hold — thread session owner busy; awaiting its return"
+                );
+                if let Some(observer) = observer {
+                    observer.emit(
+                        "busy_owner_hold",
+                        None,
+                        &observer::context_for(Some(channel_id), None, None),
+                        serde_json::json!({
+                            "scope": scope.telemetry_label(),
+                            "ownerIndex": owner_index,
+                            "heldForSecs": held_for.as_secs_f64(),
+                            "timeoutSecs": pool::HOLD_BUSY_OWNER_TIMEOUT.as_secs_f64(),
+                        }),
+                    );
+                }
+                held.push(batch);
+                continue;
+            }
+            pool::HoldDecision::ForkAfterHold {
+                held_for,
+                owner_index,
+            } => {
+                tracing::warn!(
+                    channel = %channel_id,
+                    scope = %scope.telemetry_label(),
+                    owner_index,
+                    held_for_secs = held_for.as_secs_f64(),
+                    "busy-owner hold expired — forking fresh session on an idle worker"
+                );
+                if let Some(observer) = observer {
+                    observer.emit(
+                        "busy_owner_hold_forked",
+                        None,
+                        &observer::context_for(Some(channel_id), None, None),
+                        serde_json::json!({
+                            "scope": scope.telemetry_label(),
+                            "ownerIndex": owner_index,
+                            "heldForSecs": held_for.as_secs_f64(),
+                        }),
+                    );
+                }
+                // Fall through to try_claim below (fork); record_scope_owner
+                // reassigns ownership to the new worker automatically.
+            }
+            pool::HoldDecision::Dispatch => {}
         }
         let typing_scope = batch
             .events
@@ -6117,13 +6197,61 @@ mod owner_control_command_tests {
         // A brand-new thread with no recorded owner is never held.
         assert!(!pool.should_hold_for_busy_owner(&thread_scope(ch, &"d".repeat(64))));
 
-        // Channel-wide session invalidation prunes the directory so a stale
-        // owner can never strand a held batch.
+        // The bounded hold decision stamps A's first-held time, then forks once
+        // the window elapses rather than starving behind the busy owner.
+        let now = std::time::Instant::now();
+        assert!(
+            matches!(
+                pool.hold_decision(&ta, now, pool::HOLD_BUSY_OWNER_TIMEOUT),
+                pool::HoldDecision::Hold { .. }
+            ),
+            "busy owner within window => hold"
+        );
+        assert!(pool.held_since_contains(&ta), "hold stamps first-held time");
+        assert!(
+            matches!(
+                pool.hold_decision(
+                    &ta,
+                    now + pool::HOLD_BUSY_OWNER_TIMEOUT,
+                    pool::HOLD_BUSY_OWNER_TIMEOUT
+                ),
+                pool::HoldDecision::ForkAfterHold { .. }
+            ),
+            "elapsed window => fork on an idle worker"
+        );
+        assert!(
+            !pool.held_since_contains(&ta),
+            "fork clears the first-held stamp"
+        );
+
+        // A conversation scope never holds even with a busy recorded owner —
+        // this is the cross-channel head-of-line-blocking regression guard.
+        let cs = scope::SessionScope::Conversation { channel_id: ch };
+        pool.record_scope_owner(cs.clone(), 0);
+        assert_eq!(
+            pool.hold_decision(&cs, now, pool::HOLD_BUSY_OWNER_TIMEOUT),
+            pool::HoldDecision::Dispatch,
+            "conversation scope forks a busy owner rather than holding"
+        );
+
+        // Re-stamp A's hold so channel invalidation has an entry to prune.
+        assert!(matches!(
+            pool.hold_decision(&ta, now, pool::HOLD_BUSY_OWNER_TIMEOUT),
+            pool::HoldDecision::Hold { .. }
+        ));
+        assert!(pool.held_since_contains(&ta));
+
+        // Channel-wide session invalidation prunes the owner directory and the
+        // hold stamps so a stale owner can never strand a held batch.
         pool.invalidate_channel_sessions(ch);
         assert!(
             !pool.should_hold_for_busy_owner(&ta),
             "owner directory pruned on channel invalidation"
         );
+        assert!(
+            !pool.held_since_contains(&ta),
+            "hold stamps pruned on channel invalidation"
+        );
     }
 
     #[test]

crates/buzz-acp/src/pool.rs

@@ -337,6 +337,12 @@ pub struct AgentPool {
     /// Best-effort: stale entries (rotation, crash/respawn) self-heal on the
     /// next dispatch and are pruned on channel-wide session invalidation.
     session_owners: HashMap<SessionScope, usize>,
+    /// First time each scope was held for a busy owner, so the bounded hold can
+    /// expire and fork rather than starve behind an unbounded turn. Derived
+    /// state: cleared on every dispatch/invalidation path, and only ever holds
+    /// `Thread` scopes (the sole variant [`hold_decision`](Self::hold_decision)
+    /// stamps).
+    held_since: HashMap<SessionScope, std::time::Instant>,
 }
 
 /// Result returned by a completed prompt task.
@@ -824,6 +830,7 @@ impl AgentPool {
             join_set: JoinSet::new(),
             task_map: HashMap::new(),
             session_owners: HashMap::new(),
+            held_since: HashMap::new(),
         }
     }
 
@@ -852,6 +859,44 @@ impl AgentPool {
         }
     }
 
+    /// Decide whether to hold `scope`'s batch for its busy session owner, fork it
+    /// after a bounded hold, or dispatch immediately. Stamps and clears the
+    /// first-held time internally so the bounded window survives across dispatch
+    /// cycles without a dedicated timer; `now` and `timeout` are injected for
+    /// testability.
+    ///
+    /// Gated on the scope variant, not the session policy: `Conversation` scopes
+    /// (channel-policy channels and all DMs) never hold — a busy owner there means
+    /// fork onto another idle worker, the pre-thread-sessions behavior. Only
+    /// `Thread` scopes hold, so a momentarily busy owner does not cause a
+    /// duplicate provider session for the same thread.
+    pub fn hold_decision(
+        &mut self,
+        scope: &SessionScope,
+        now: std::time::Instant,
+        timeout: Duration,
+    ) -> HoldDecision {
+        if !scope.is_thread() || !self.should_hold_for_busy_owner(scope) {
+            self.held_since.remove(scope);
+            return HoldDecision::Dispatch;
+        }
+        let owner_index = self.session_owners.get(scope).copied().unwrap_or_default();
+        let first = *self.held_since.entry(scope.clone()).or_insert(now);
+        let held_for = now.saturating_duration_since(first);
+        if held_for >= timeout {
+            self.held_since.remove(scope);
+            HoldDecision::ForkAfterHold {
+                held_for,
+                owner_index,
+            }
+        } else {
+            HoldDecision::Hold {
+                held_for,
+                owner_index,
+            }
+        }
+    }
+
     /// Try to claim an idle agent for the given session scope (or heartbeat if
     /// `None`).
     ///
@@ -927,6 +972,13 @@ impl AgentPool {
         &mut self.task_map
     }
 
+    /// Whether a first-held stamp is currently recorded for `scope`. Test seam
+    /// for [`hold_decision`](Self::hold_decision) callers outside this module.
+    #[cfg(test)]
+    pub(crate) fn held_since_contains(&self, scope: &SessionScope) -> bool {
+        self.held_since.contains_key(scope)
+    }
+
     /// Try to send a goose-native steer request to the in-flight task for
     /// `channel_id`.
     ///
@@ -1057,6 +1109,10 @@ impl AgentPool {
         // owner after the channel's sessions are gone.
         self.session_owners
             .retain(|scope, _| scope.channel_id() != channel_id);
+        // Prune held-since stamps for the same channel so an expiring hold cannot
+        // reference a scope whose sessions are gone.
+        self.held_since
+            .retain(|scope, _| scope.channel_id() != channel_id);
         count
     }
 
@@ -1078,6 +1134,7 @@ impl AgentPool {
             }
         }
         self.session_owners.remove(scope);
+        self.held_since.remove(scope);
         count
     }
 
@@ -1164,10 +1221,31 @@ impl AgentPool {
         agent.desired_model_request_id = request_id;
         agent.state.invalidate_scope(&scope);
         self.session_owners.remove(&scope);
+        self.held_since.remove(&scope);
         IdleSwitchResult::Switched
     }
 }
 
+/// Outcome of [`AgentPool::hold_decision`] for one queued batch.
+#[derive(Debug, PartialEq, Eq)]
+pub enum HoldDecision {
+    /// Dispatch now: never-hold scope (conversation), idle owner holds the
+    /// session, or no busy owner is recorded.
+    Dispatch,
+    /// Leave queued this cycle: the thread's session owner is busy and the
+    /// bounded hold window has not elapsed.
+    Hold {
+        held_for: Duration,
+        owner_index: usize,
+    },
+    /// Bounded hold expired — dispatch anyway, forking a fresh session on an
+    /// idle worker.
+    ForkAfterHold {
+        held_for: Duration,
+        owner_index: usize,
+    },
+}
+
 /// Outcome of [`AgentPool::switch_idle_agent_model`].
 #[derive(Debug, PartialEq, Eq)]
 pub enum IdleSwitchResult {
@@ -1206,6 +1284,12 @@ const CONTROL_CANCEL_GRACE: Duration = Duration::from_secs(5);
 /// Timeout for permission-mode requests (`session/set_config_option` with `configId: "mode"`).
 const PERMISSION_MODE_TIMEOUT: Duration = Duration::from_secs(5);
 
+/// Bounded window a `Thread` batch waits for its busy session-owner before we
+/// stop holding and fork a fresh session on an idle worker. Kept below the 30s
+/// maintenance tick so even a silent system re-evaluates a held batch shortly
+/// after expiry, versus the max-turn deadline it could starve behind today.
+pub(crate) const HOLD_BUSY_OWNER_TIMEOUT: Duration = Duration::from_secs(10);
+
 /// Placeholder [`fetch_channel_info`] substitutes when a channel's metadata
 /// event carries no `name` tag. Not a real channel name — consumers that need
 /// an identifying name must treat it as absent.
@@ -7379,6 +7463,9 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
         let mut pool = AgentPool::from_slots(vec![Some(agent)]);
         pool.record_scope_owner(ta.clone(), 0);
         pool.record_scope_owner(tb.clone(), 0);
+        let now = std::time::Instant::now();
+        pool.held_since.insert(ta.clone(), now);
+        pool.held_since.insert(tb.clone(), now);
 
         let cleared = pool.invalidate_scope_session(&ta);
 
@@ -7396,6 +7483,186 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
             pool.session_owners.contains_key(&tb),
             "thread B owner retained"
         );
+        assert!(
+            !pool.held_since.contains_key(&ta),
+            "thread A hold stamp dropped"
+        );
+        assert!(
+            pool.held_since.contains_key(&tb),
+            "thread B hold stamp retained"
+        );
+    }
+
+    /// Insert a `task_map` entry so `agent_index` reads as checked-out (busy)
+    /// for the busy-owner predicate, mirroring an in-flight prompt task without
+    /// spawning a real one. `busy_scope` is the turn the worker is running.
+    fn mark_agent_busy(pool: &mut AgentPool, agent_index: usize, busy_scope: SessionScope) {
+        let abort = pool.join_set.spawn(async {});
+        pool.task_map_mut().insert(
+            abort.id(),
+            TaskMeta {
+                agent_index,
+                channel_id: Some(busy_scope.channel_id()),
+                scope: Some(busy_scope),
+                turn_id: "t".into(),
+                recoverable_batch: None,
+                control_tx: None,
+                steer_tx: None,
+                successful_steer_deliveries: HashSet::new(),
+            },
+        );
+    }
+
+    /// An idle agent (slot 0) holding a provider session for `scope`, so
+    /// `has_session_for(scope)` is true.
+    async fn idle_agent_with_session(scope: SessionScope) -> OwnedAgent {
+        let acp = AcpClient::spawn("bash", &["-c".into(), "sleep 10".into()], &[], false)
+            .await
+            .expect("spawn dummy ACP");
+        let mut agent = OwnedAgent {
+            index: 0,
+            acp,
+            state: SessionState::default(),
+            model_capabilities: None,
+            desired_model: None,
+            model_overridden: false,
+            desired_model_request_id: None,
+            desired_model_pending_ack: false,
+            startup_effort: None,
+            agent_name: "test".into(),
+            goose_system_prompt_supported: None,
+            protocol_version: 2,
+        };
+        agent.state.sessions.insert(scope, "sess".into());
+        agent
+    }
+
+    // `hold_decision` is gated on the scope variant (not session policy),
+    // short-circuits when an idle worker already holds the session or no busy
+    // owner is recorded, and only a busy `Thread` owner holds — for a bounded
+    // window, after which it forks. The `Conversation` + busy row is the
+    // cross-channel head-of-line-blocking regression guard (PR #6732).
+    #[tokio::test]
+    async fn hold_decision_covers_variant_session_busy_and_timeout() {
+        #[derive(Debug)]
+        enum Expect {
+            Dispatch,
+            Hold,
+            ForkAfterHold,
+        }
+        struct Row {
+            name: &'static str,
+            is_thread: bool,
+            has_session: bool,
+            owner_busy: bool,
+            elapsed: Duration,
+            expect: Expect,
+        }
+        let timeout = Duration::from_secs(10);
+        let rows = [
+            // Cross-channel regression guard: a conversation scope with a busy
+            // recorded owner dispatches (forks) rather than starving a sibling.
+            Row {
+                name: "conversation + busy owner dispatches",
+                is_thread: false,
+                has_session: false,
+                owner_busy: true,
+                elapsed: Duration::ZERO,
+                expect: Expect::Dispatch,
+            },
+            // An idle worker already holds the thread session — reuse it.
+            Row {
+                name: "thread + idle session dispatches",
+                is_thread: true,
+                has_session: true,
+                owner_busy: true,
+                elapsed: Duration::ZERO,
+                expect: Expect::Dispatch,
+            },
+            // No busy owner recorded — nothing to wait for.
+            Row {
+                name: "thread + no busy owner dispatches",
+                is_thread: true,
+                has_session: false,
+                owner_busy: false,
+                elapsed: Duration::ZERO,
+                expect: Expect::Dispatch,
+            },
+            // Busy thread owner within the window — hold.
+            Row {
+                name: "thread + busy owner within window holds",
+                is_thread: true,
+                has_session: false,
+                owner_busy: true,
+                elapsed: Duration::ZERO,
+                expect: Expect::Hold,
+            },
+            // Busy thread owner past the window — fork onto an idle worker.
+            Row {
+                name: "thread + busy owner past window forks",
+                is_thread: true,
+                has_session: false,
+                owner_busy: true,
+                elapsed: timeout,
+                expect: Expect::ForkAfterHold,
+            },
+        ];
+
+        let base = std::time::Instant::now();
+        for row in rows {
+            let ch = Uuid::new_v4();
+            let scope = if row.is_thread {
+                thread_scope(ch, &"a".repeat(64))
+            } else {
+                conv(ch)
+            };
+            let slots = if row.has_session {
+                vec![Some(idle_agent_with_session(scope.clone()).await)]
+            } else {
+                vec![]
+            };
+            let mut pool = AgentPool::from_slots(slots);
+            if row.owner_busy {
+                pool.record_scope_owner(scope.clone(), 1);
+                mark_agent_busy(&mut pool, 1, thread_scope(ch, &"b".repeat(64)));
+            }
+
+            // A non-zero elapsed needs a first stamping call before the second
+            // evaluates the window against the same base instant.
+            if !row.elapsed.is_zero() {
+                assert!(
+                    matches!(
+                        pool.hold_decision(&scope, base, timeout),
+                        HoldDecision::Hold { .. }
+                    ),
+                    "{}: first call stamps a hold",
+                    row.name
+                );
+            }
+            let decision = pool.hold_decision(&scope, base + row.elapsed, timeout);
+
+            match (&row.expect, &decision) {
+                (Expect::Dispatch, HoldDecision::Dispatch)
+                | (Expect::Hold, HoldDecision::Hold { .. })
+                | (Expect::ForkAfterHold, HoldDecision::ForkAfterHold { .. }) => {}
+                _ => panic!("{}: expected {:?}, got {decision:?}", row.name, row.expect),
+            }
+
+            // held_since holds the scope only while a Hold is outstanding.
+            if matches!(decision, HoldDecision::Hold { .. }) {
+                assert!(
+                    pool.held_since.contains_key(&scope),
+                    "{}: hold stamps held_since",
+                    row.name
+                );
+            } else {
+                assert!(
+                    !pool.held_since.contains_key(&scope),
+                    "{}: dispatch/fork clears held_since",
+                    row.name
+                );
+            }
+        }
     }
 
     #[test]
@@ -10172,6 +10439,8 @@ done"#
         // selected scope and its owner are cleared without broad channel cleanup.
         pool.invalidate_scope_session(&scopes[1]);
         pool.record_scope_owner(scopes[0].clone(), 0);
+        pool.held_since
+            .insert(scopes[0].clone(), std::time::Instant::now());
         assert_eq!(
             pool.switch_idle_agent_model(channel_id, "model-b", Some("pick".into())),
             IdleSwitchResult::Switched,
@@ -10180,6 +10449,10 @@ done"#
         assert_eq!(agent.desired_model.as_deref(), Some("model-b"));
         assert!(!agent.state.sessions.contains_key(&scopes[0]));
         assert!(!pool.session_owners.contains_key(&scopes[0]));
+        assert!(
+            !pool.held_since.contains_key(&scopes[0]),
+            "switched scope's hold stamp cleared with its session"
+        );
     }
 
     #[tokio::test]