Add CapturingOutputSink and worker correctness unit tests

Adds a CapturingOutputSink to output_sink.rs that stores all emitted
(PrecomputedOutput, Box<dyn AggregateCore>) pairs in a Mutex<Vec> for
inspection in tests, alongside drain() and len() helpers.

Adds 6 unit tests in worker.rs covering:
- Raw mode: each sample forwarded as SumAccumulator with sum == value
- Tumbling window: correct boundary and aggregated sum on window close
- Sliding window pane sharing: single sample emits in both overlapping
  windows with the correct sum via snapshot/take pane merge
- GROUP BY separate emits: two series on the same worker produce
  independent per-series accumulators (no ingest-time cross-series merge)
- Late data Drop: sample behind watermark - allowed_lateness not emitted
- Late data ForwardToStore: late sample for evicted pane emitted as
  mini-accumulator with correct window bounds and sum

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: zzylol

asap-query-engine/Cargo.toml

@@ -63,6 +63,7 @@ path = "src/bin/precompute_engine.rs"
 name = "test_e2e_precompute"
 path = "src/bin/test_e2e_precompute.rs"
 
+
 [dev-dependencies]
 tempfile = "3.20.0"
 

asap-query-engine/src/precompute_engine/output_sink.rs

@@ -1,6 +1,6 @@
 use crate::data_model::{AggregateCore, PrecomputedOutput};
 use crate::stores::Store;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use tracing::debug_span;
 
 /// Trait for emitting completed window outputs.
@@ -61,6 +61,43 @@ impl OutputSink for RawPassthroughSink {
     }
 }
 
+/// A capturing sink for testing that stores all emitted outputs.
+pub struct CapturingOutputSink {
+    pub captured: Mutex<Vec<(PrecomputedOutput, Box<dyn AggregateCore>)>>,
+}
+
+impl CapturingOutputSink {
+    pub fn new() -> Self {
+        Self {
+            captured: Mutex::new(Vec::new()),
+        }
+    }
+
+    pub fn drain(&self) -> Vec<(PrecomputedOutput, Box<dyn AggregateCore>)> {
+        self.captured.lock().unwrap().drain(..).collect()
+    }
+
+    pub fn len(&self) -> usize {
+        self.captured.lock().unwrap().len()
+    }
+}
+
+impl Default for CapturingOutputSink {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl OutputSink for CapturingOutputSink {
+    fn emit_batch(
+        &self,
+        outputs: Vec<(PrecomputedOutput, Box<dyn AggregateCore>)>,
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        self.captured.lock().unwrap().extend(outputs);
+        Ok(())
+    }
+}
+
 /// A no-op sink for testing that just counts emitted batches.
 pub struct NoopOutputSink {
     pub emit_count: std::sync::atomic::AtomicU64,