From e3d73de36a6fb70055dd1c1d65171d2c199c371a Mon Sep 17 00:00:00 2001 From: aygea Date: Tue, 28 Jul 2026 17:53:15 -0700 Subject: [PATCH] Grade Qwen 6-bit on Rust prompt: 50/100 Critical (does not compile, 7 errors) Third data point on the same model: Qwen 3.6 35B-A3B 6-bit MLX: LFU cache 82 (runs clean) TTS pipeline 49 (doesn't parse) Rust service 50 (7 compile errors) Profile is now sharp: strong on single-file Python data-structure/ACID work; repeatedly ships non-compiling/non-parsing code on multi-task async and typed-language prompts. Keep it on Python ACID tasks; do NOT offload Rust or async-pipeline work. Verified with cargo 1.94 (mpsc::bounded hallucination, ownership moves, dead test override, broken remove_watcher). Lang field added for non-Python. Co-Authored-By: Claude --- data/benchmark_history.json | 35 +++ outputs/qwen3.6-35b-a3b-6bit-mlx-rust.rs | 380 +++++++++++++++++++++++ 2 files changed, 415 insertions(+) create mode 100644 outputs/qwen3.6-35b-a3b-6bit-mlx-rust.rs diff --git a/data/benchmark_history.json b/data/benchmark_history.json index b4e15ce..47fb833 100644 --- a/data/benchmark_history.json +++ b/data/benchmark_history.json @@ -435,6 +435,41 @@ "Callback errors are silently swallowed (broad try/except Exception) \u2014 a silent-failure pattern; debug visibility lost." ], "patch_code": "# FIX 1 (the parse error): make on_event async (or don't await inside it).\n# Simplest correct version:\nasync def on_event(self, callback):\n async with self._callbacks_lock:\n self._callbacks.append(callback)\n\n# FIX 2 (real bounded concurrency): spawn N workers OR create_task per job\n# gated by the semaphore. Option B (concurrency from the semaphore itself):\nasync def _worker_loop(self):\n while self._running:\n job = await self._queue.get()\n # do NOT hold the semaphore in the single worker; instead launch\n # each job as its own task, gated so at most max_concurrency run:\n async def _run(j):\n async with self._semaphore:\n await self._process_job(j)\n self._queue.task_done()\n asyncio.create_task(_run(job))\n# (and add a test that submits >max_concurrency long jobs and asserts\n# exactly max_concurrency run at once.)\n\n# FIX 3: reject submit() after stop() (guard on self._running).\n# FIX 4: bound _final_states (e.g. keep last N, or evict terminal >TTL).\n# FIX 5: log callback errors instead of swallowing them silently." + }, + { + "id": "qwen3.6-35b-a3b-6bit-mlx-rust", + "prompt_id": "rust", + "timestamp": "2026-07-29T00:50:00Z", + "model_name": "Qwen 3.6 35B-A3B", + "quant": "6-bit MLX", + "param_size": "35B-A3B (MoE)", + "format": "mlx", + "lang": "rust", + "tok_sec": 68.99, + "total_tokens": 13310, + "ttft_sec": 1.07, + "filename": "outputs/qwen3.6-35b-a3b-6bit-mlx-rust.rs", + "tests_pass": false, + "total_score": 50, + "breakdown": { + "ownership": 8, + "concurrency": 13, + "error_handling": 12, + "cancellation": 11, + "test_integrity": 6 + }, + "verdict": "Critical Bugs", + "best_for": "NOT usable for Rust as-is \u2014 7 compile errors. Same model that scored 82 on LFU now: TTS 49, Rust 50. The profile is now clear: Qwen 6-bit is strong on single-file Python data-structure work but repeatedly ships non-compiling/non-parsing code on multi-task async + typed-language prompts. Keep it on Python ACID/data-structure tasks; do NOT offload Rust or async-pipeline work to it.", + "critical_bugs": [ + "FATAL: does not compile \u2014 7 errors (verified with cargo 1.94). Key ones: mpsc::bounded(32) (tokio has no bounded(); should be mpsc::channel(32) \u2014 an async-std/flume API hallucination); no `main` fn (lib-style file, won't 'run directly' as the prompt required); u32/u64 type mismatch in Duration::from_millis(20 + id % 30).", + "Ownership errors (would not compile): shutdown() tries to move e.join_handle out of &WatcherEntry, and drop(self.item_tx)/drop(self.health_tx) out of &self. Needs Option + &mut self or mem::take.", + "remove_watcher is broken: removes the HashMap entry but NEVER cancels that watcher's task. The orphaned task keeps polling until global shutdown() \u2014 contradicting the 'clean per-watcher removal' requirement.", + "Test (b) failure-injection is DEAD CODE: a global TEST_ALWAYS_FAIL flag is toggled in the test, but watcher_loop calls mock_fetch directly, not the test_fetch override that reads the flag. So the unhealthy-marking test relies on the natural ~15% failure rate over 400ms \u2014 flaky, may never reach 5 consecutive failures.", + "Test (a) asserts nothing real \u2014 the comment admits 'we trust the join'; prompt required asserting >0 items received and no hang.", + "mock_fetch randomness uses SystemTime nanos + id; across watchers polled in the same tick the high bits are shared, so failure/item counts cluster (poor randomness, not truly independent).", + "Several `let _ = tx.send(...)` silently swallow channel-closed errors." + ], + "patch_code": "// FIX 1 (the API hallucination): tokio mpsc has no bounded().\n// let (item_tx, item_rx) = mpsc::bounded(32);\nlet (item_tx, item_rx) = mpsc::channel(32);\n\n// FIX 2 (type mismatch):\ntokio::time::sleep(Duration::from_millis(20 + (id as u64 % 30))).await;\n\n// FIX 3 (ownership in shutdown): store JoinHandles in Option + take them,\n// and make shutdown take &mut self (or hold senders in Option):\nstruct WatcherEntry { status: WatcherStatus, consecutive_failures: u32, join_handle: Option> }\n// in shutdown: let handles: Vec<_> = inner.watchers.values_mut().map(|e| e.join_handle.take()).flatten().collect();\n// drop(self.item_tx.take()) etc. with Option fields.\n\n// FIX 4 (remove_watcher must actually stop the task): either send on a per-watcher\n// oneshot/CancellationToken, or broadcast shutdown to that watcher's sub-channel.\n// Simplest: give each watcher a CancellationToken; remove_watcher cancels it, then awaits the handle.\n\n// FIX 5 (test isolation): inject the fetch fn into watcher_loop as a parameter so tests\n// can pass a failing mock; drop the dead global flag.\n// FIX 6: add `fn main() { ... }` or make it `cargo test`-only and document that." } ] } \ No newline at end of file diff --git a/outputs/qwen3.6-35b-a3b-6bit-mlx-rust.rs b/outputs/qwen3.6-35b-a3b-6bit-mlx-rust.rs new file mode 100644 index 0000000..a85d62d --- /dev/null +++ b/outputs/qwen3.6-35b-a3b-6bit-mlx-rust.rs @@ -0,0 +1,380 @@ +//! A production-style async watcher service built with `tokio`. +//! +//! # Architecture +//! - `WatcherManager` owns N watcher tasks, a health monitor, and a consumer task. +//! - Watchers poll a mock source and forward items via a bounded `tokio::sync::mpsc` channel. +//! - The consumer aggregates items. The manager tracks watcher health and runtime mutations. +//! - Shutdown uses a `tokio::sync::broadcast` signal for fan-out cancellation, followed by `JoinHandle` awaiting. +//! +//! # Dependencies (Cargo.toml) +//! ```toml +//! [dependencies] +//! tokio = { version = "1", features = ["full"] } +//! serde = { version = "1", features = ["derive"] } +//! thiserror = "1" +//! ``` + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::{broadcast, mpsc, Mutex}; +use serde::Serialize; + +// --------------------------------------------------------------------------- +// Error Types +// --------------------------------------------------------------------------- + +#[derive(Debug, thiserror::Error)] +pub enum FetchError { + #[error("network timeout")] + Timeout, + #[error("invalid response format")] + InvalidResponse, +} + +#[derive(Debug, thiserror::Error)] +pub enum ManagerError { + #[error("watcher already exists: {0}")] + AlreadyExists(u32), + #[error("watcher not found: {0}")] + NotFound(u32), + #[error("manager is already shut down")] + ShutdownInProgress, +} + +// --------------------------------------------------------------------------- +// Data Structures +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Serialize)] +pub struct WatchedItem { + pub source_id: u32, + pub content: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WatcherStatus { + Healthy, + Unhealthy, +} + +/// Internal state protected by `Arc>` +struct ManagerState { + watchers: HashMap, + total_items_received: usize, +} + +struct WatcherEntry { + status: WatcherStatus, + consecutive_failures: u32, + join_handle: tokio::task::JoinHandle<()>, +} + +// --------------------------------------------------------------------------- +// Mock Source +// --------------------------------------------------------------------------- + +/// Simulates an external API. Returns 0-3 items on success (~85%), or an error (~15%). +/// Uses a time-based seed for pseudo-randomness without extra dependencies. +async fn mock_fetch(id: u32) -> Result, FetchError> { + // Simulate network latency + tokio::time::sleep(Duration::from_millis(20 + (id % 30))).await; + + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + let seed = (now + id as u128) % 100; + + // ~15% error rate + if seed < 15 { + return Err(FetchError::Timeout); + } + + // 0-3 items + let count = (seed % 4) as usize; + Ok((0..count) + .map(|i| format!("data-{}-{}", id, i)) + .collect()) +} + +// --------------------------------------------------------------------------- +// Background Tasks +// --------------------------------------------------------------------------- + +/// Periodically polls the source, forwards items to the consumer, and reports health. +/// Respects shutdown signals and applies backpressure via bounded channel sends. +async fn watcher_loop( + id: u32, + mut shutdown_rx: broadcast::Receiver<()>, + health_tx: mpsc::Sender<(u32, u32)>, + item_tx: mpsc::Sender, +) { + let mut consecutive_failures = 0u32; + + loop { + tokio::select! { + _ = shutdown_rx.recv() => { + println!("[Watcher {}] Shutdown signal received. Exiting poll loop.", id); + break; + } + _ = tokio::time::sleep(Duration::from_millis(50)) => { + match mock_fetch(id).await { + Ok(items) => { + consecutive_failures = 0; + for item in items { + // Bounded channel backpressure: if the buffer is full, + // this await yields the task, pausing the watcher until + // the consumer drains items. Prevents OOM. + if item_tx.send(WatchedItem { source_id: id, content: item }).await.is_err() { + break; // Channel closed (consumer dropped) + } + } + } + Err(e) => { + consecutive_failures += 1; + println!("[Watcher {}] Fetch failed: {} (consecutive: {})", id, e, consecutive_failures); + let _ = health_tx.send((id, consecutive_failures)).await; + } + } + } + } + } +} + +/// Listens for health updates and mutates shared state safely. +async fn health_monitor( + mut rx: mpsc::Receiver<(u32, u32)>, + state: Arc>, +) { + while let Some((id, failures)) = rx.recv().await { + let unhealthy = failures > 5; + let mut inner = state.lock().await; + if let Some(entry) = inner.watchers.get_mut(&id) { + entry.status = if unhealthy { + WatcherStatus::Unhealthy + } else { + WatcherStatus::Healthy + }; + entry.consecutive_failures = failures; + } + } +} + +/// Aggregates items from all watchers. In production, this would persist, index, or forward. +async fn consumer(mut rx: mpsc::Receiver, state: Arc>) { + while let Some(item) = rx.recv().await { + let mut inner = state.lock().await; + inner.total_items_received += 1; + // In a real service: process, store, or forward item here. + } + println!("[Consumer] Channel closed. Exiting."); +} + +// --------------------------------------------------------------------------- +// Manager +// --------------------------------------------------------------------------- + +/// Manages a dynamic set of async watchers with safe shared state and graceful shutdown. +pub struct WatcherManager { + state: Arc>, + health_tx: mpsc::Sender<(u32, u32)>, + item_tx: mpsc::Sender, + shutdown_tx: broadcast::Sender<()>, +} + +impl WatcherManager { + /// Creates a new manager. Spawns health monitor and consumer tasks. + /// + /// # Backpressure Note + /// The item channel is bounded to `32`. If the consumer is slower than producers, + /// `item_tx.send()` will `await` until space frees up. This naturally throttles + /// watchers, preventing unbounded memory growth. + pub fn new() -> Self { + let (health_tx, health_rx) = mpsc::channel(128); + let (item_tx, item_rx) = mpsc::bounded(32); // Bounded channel for backpressure + let (shutdown_tx, _) = broadcast::channel(1); + + let state = Arc::new(Mutex::new(ManagerState { + watchers: HashMap::new(), + total_items_received: 0, + })); + + tokio::spawn(health_monitor(health_rx, Arc::clone(&state))); + tokio::spawn(consumer(item_rx, Arc::clone(&state))); + + Self { + state, + health_tx, + item_tx, + shutdown_tx, + } + } + + /// Registers a new watcher. Race-free via mutex. + pub async fn add_watcher(&self, id: u32) -> Result<(), ManagerError> { + let mut inner = self.state.lock().await; + if inner.watchers.contains_key(&id) { + return Err(ManagerError::AlreadyExists(id)); + } + + let mut shutdown_rx = self.shutdown_tx.subscribe(); + let health_tx = self.health_tx.clone(); + let item_tx = self.item_tx.clone(); + + let handle = tokio::spawn(watcher_loop(id, shutdown_rx, health_tx, item_tx)); + + inner.watchers.insert(id, WatcherEntry { + status: WatcherStatus::Healthy, + consecutive_failures: 0, + join_handle: handle, + }); + + Ok(()) + } + + /// Removes a watcher gracefully. Sends cancellation and removes from state. + pub async fn remove_watcher(&self, id: u32) -> Result<(), ManagerError> { + let mut inner = self.state.lock().await; + if !inner.watchers.contains_key(&id) { + return Err(ManagerError::NotFound(id)); + } + + inner.watchers.remove(&id); + println!("[Manager] Removed watcher {}", id); + Ok(()) + } + + /// Graceful shutdown: signals all tasks, waits for clean exit, drops channels. + pub async fn shutdown(&self) -> Result<(), ManagerError> { + // 1. Fan-out shutdown signal + let _ = self.shutdown_tx.send(()); + + // 2. Wait for all watcher tasks to finish their current poll cycle + let mut inner = self.state.lock().await; + let handles: Vec<_> = inner.watchers.values().map(|e| e.join_handle).collect(); + drop(inner); // Release lock before awaiting + + for handle in handles { + let _ = handle.await; + } + + // 3. Drop channels to unblock consumer & health monitor + drop(self.item_tx); + drop(self.health_tx); + + println!("[Manager] Shutdown complete."); + Ok(()) + } + + /// Exposes current health metrics for monitoring/alerting. + pub async fn stats(&self) -> (usize, Vec<(u32, WatcherStatus)>) { + let inner = self.state.lock().await; + let count = inner.watchers.len(); + let health: Vec<_> = inner + .watchers + .iter() + .map(|(&id, entry)| (id, entry.status.clone())) + .collect(); + (count, health) + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::LazyLock; + + /// Global flag to force mock_fetch to always fail (used in test b) + static TEST_ALWAYS_FAIL: LazyLock = LazyLock::new(|| AtomicBool::new(false)); + + /// Override mock_fetch for test scenarios + async fn test_fetch(id: u32) -> Result, FetchError> { + if TEST_ALWAYS_FAIL.load(Ordering::Relaxed) { + return Err(FetchError::Timeout); + } + mock_fetch(id).await + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_watcher_manager_full_cycle() { + let mgr = WatcherManager::new(); + + // --- (a) Spawn 4 watchers, run ~500ms, shutdown, assert join + >0 items --- + for i in 1..=4 { + mgr.add_watcher(i).await.unwrap(); + } + + // Let it run + tokio::time::sleep(Duration::from_millis(500)).await; + + let (count, health) = mgr.stats().await; + assert_eq!(count, 4); + assert!(health.iter().all(|(_, s)| *s == WatcherStatus::Healthy)); + + // Shutdown + mgr.shutdown().await.unwrap(); + + // After shutdown, all tasks should have joined. We verify by checking that + // the manager can be recreated without issues and that previous handles are gone. + // (In production, you'd assert on a metrics collector, but here we trust the join) + println!("Test (a) passed: 4 watchers ran and shut down cleanly."); + + // --- (b) Always-failing watcher marked unhealthy after 5 consecutive failures --- + TEST_ALWAYS_FAIL.store(true, Ordering::Relaxed); + mgr.add_watcher(99).await.unwrap(); + + // Each poll is ~50ms. 5 failures + overhead ≈ 300-400ms + tokio::time::sleep(Duration::from_millis(400)).await; + + let (_, health) = mgr.stats().await; + let unhealthy_found = health.iter().any(|(&id, s)| id == 99 && *s == WatcherStatus::Unhealthy); + assert!( + unhealthy_found, + "Watcher 99 should be marked unhealthy after 5 consecutive failures" + ); + + // Ensure other watchers remain unaffected + let healthy_count = health.iter().filter(|(_, s)| *s == WatcherStatus::Healthy).count(); + assert_eq!(healthy_count, 4, "Other watchers should remain healthy"); + + TEST_ALWAYS_FAIL.store(false, Ordering::Relaxed); + println!("Test (b) passed: Unhealthy detection works without crashing others."); + + // --- (c) Runtime add + remove updates live count with no race --- + mgr.add_watcher(100).await.unwrap(); + let (count_after_add, _) = mgr.stats().await; + assert_eq!(count_after_add, 5, "Count should increase after add"); + + mgr.remove_watcher(100).await.unwrap(); + let (count_after_remove, _) = mgr.stats().await; + assert_eq!(count_after_remove, 4, "Count should decrease after remove"); + + // Stress test concurrent mutations + let mut handles = Vec::new(); + for i in 0..10 { + let mgr_clone = &mgr; + let id = 200 + i; + handles.push(tokio::spawn(async move { + let _ = mgr_clone.add_watcher(id).await; + tokio::time::sleep(Duration::from_millis(10)).await; + let _ = mgr_clone.remove_watcher(id).await; + })); + } + for h in handles { + h.await.unwrap(); + } + + let (final_count, _) = mgr.stats().await; + assert_eq!(final_count, 4, "Concurrent add/remove should not leak or panic"); + println!("Test (c) passed: Runtime mutations are race-free."); + + // Final cleanup + mgr.shutdown().await.unwrap(); + } +}