From e470d235764b99213fee759a90838fae29104133 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E5=BE=B7=E5=AE=87?= Date: Tue, 15 Sep 2026 20:00:47 +0800 Subject: [PATCH] =?UTF-8?q?=E9=81=BF=E5=85=8D=E8=AF=AF=E6=B7=98=E6=B1=B0?= =?UTF-8?q?=E5=AE=9E=E6=97=B6=20DirectThread=20=E8=AE=A2=E9=98=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 仅在订阅游标落后于队列头时驱逐订阅，并补充队头被 pin 的回归测试。 --- .../src/agent/direct_thread_manager.rs | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_thread_manager.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_thread_manager.rs index b2701dba8..a2465bab7 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_thread_manager.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_thread_manager.rs @@ -364,9 +364,15 @@ impl DirectThreadManager { let Some(thread) = self.threads.get_mut(thread_id) else { return; }; + let oldest_seq = thread + .events + .get(thread.head) + .map(|stored| stored.event.seq) + .unwrap_or(thread.next_seq.saturating_add(1)); let slowest = thread .subscribers .iter() + .filter(|(_, subscriber)| subscriber.cursor < oldest_seq) .min_by_key(|(_, subscriber)| subscriber.cursor) .map(|(id, _)| id.clone()); if let Some(subscription_id) = slowest { @@ -530,6 +536,20 @@ mod tests { ); } + #[test] + fn current_subscriber_is_not_expired_by_pinned_queue_head() { + let mut manager = DirectThreadManager::with_limits(2, 100_000); + manager.append("thread-1", draft("item.started", "turn-1", Some("item-1"))); + let subscription = manager.subscribe("thread-1"); + manager.append("thread-1", draft("item.delta", "turn-1", Some("item-1"))); + manager.append("thread-1", draft("item.delta", "turn-1", Some("item-1"))); + manager.append("thread-1", draft("item.delta", "turn-1", Some("item-1"))); + assert_ne!( + manager.consume(&subscription.subscription_id), + Err(SUBSCRIPTION_EXPIRED.to_string()) + ); + } + #[test] fn unresolved_approval_is_kept_in_bootstrap_until_resolved() { let mut manager = DirectThreadManager::with_limits(100, 100_000);