diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2a2a79eb..01a74893 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -42,6 +42,8 @@ jobs: - run: python -m pip install -r benchmarks/requirements.txt - run: python benchmarks/bench.py validate benchmarks/smoke.jsonl - run: python -m unittest discover -s tests/benchmarks -p 'test_*.py' -v + - name: Decider validation helpers + run: python -m unittest discover -s tests/decider -p 'test_*.py' -v cuda-contract: # The CUDA backend contract and its checker. No CUDA toolkit and no GPU: both diff --git a/src/models/decider/native/Cargo.toml b/src/models/decider/native/Cargo.toml index 80573c1b..c11f694d 100644 --- a/src/models/decider/native/Cargo.toml +++ b/src/models/decider/native/Cargo.toml @@ -34,6 +34,10 @@ path = "../../../../tests/decider/contract.rs" name = "config" path = "../../../../tests/decider/config.rs" +[[test]] +name = "batching" +path = "../../../../tests/decider/batching.rs" + [[example]] name = "decider-run" path = "../../../../tests/decider/runner.rs" diff --git a/src/models/decider/native/src/batching.rs b/src/models/decider/native/src/batching.rs new file mode 100644 index 00000000..8204f639 --- /dev/null +++ b/src/models/decider/native/src/batching.rs @@ -0,0 +1,82 @@ +//! Bounded packing of independent complete rows within a single admitted request. +use anyhow::{Context, Result, ensure}; +use std::ops::Range; + +#[derive(Clone, Copy, Debug)] +pub struct BatchLimits { + max_rows: usize, + max_tokens: usize, +} +impl Default for BatchLimits { + fn default() -> Self { + Self { + max_rows: 1, + max_tokens: 4096, + } + } +} +impl BatchLimits { + pub fn new(max_rows: usize, max_tokens: usize) -> Result { + ensure!( + (1..=4).contains(&max_rows), + "batch rows must be within 1..=4" + ); + ensure!( + (1..=4096).contains(&max_tokens), + "batch tokens must be within 1..=4096" + ); + Ok(Self { + max_rows, + max_tokens, + }) + } + pub fn from_values(rows: Option<&str>, tokens: Option<&str>) -> Result { + let rows = rows + .map_or(Ok(1), str::parse) + .context("DECIDER_BATCH_MAX_ROWS must be an integer")?; + let tokens = tokens + .map_or(Ok(4096), str::parse) + .context("DECIDER_BATCH_MAX_TOKENS must be an integer")?; + Self::new(rows, tokens) + } + pub fn from_env() -> Result { + let get = |name| match std::env::var(name) { + Ok(value) => Ok(Some(value)), + Err(std::env::VarError::NotPresent) => Ok(None), + Err(e) => Err(e), + }; + let rows = get("DECIDER_BATCH_MAX_ROWS")?; + let tokens = get("DECIDER_BATCH_MAX_TOKENS")?; + Self::from_values(rows.as_deref(), tokens.as_deref()) + } + pub fn max_rows(&self) -> usize { + self.max_rows + } + pub fn max_tokens(&self) -> usize { + self.max_tokens + } + /// Retain original row order; a row above the packing token budget runs alone. + /// Total request/row admission remains the processor/executor's separate limit. + pub fn ranges(&self, lengths: &[usize]) -> Vec> { + let mut ranges = Vec::new(); + let mut start = 0; + let mut tokens = 0usize; + for (index, &length) in lengths.iter().enumerate() { + if index > start + && (index - start == self.max_rows + || tokens + .checked_add(length) + .is_none_or(|total| total > self.max_tokens)) + { + ranges.push(start..index); + start = index; + tokens = 0; + } + tokens = tokens.saturating_add(length); + } + if start < lengths.len() { + ranges.push(start..lengths.len()); + } + ranges + } +} diff --git a/src/models/decider/native/src/engine.rs b/src/models/decider/native/src/engine.rs index f843c6a4..09ae64ad 100644 --- a/src/models/decider/native/src/engine.rs +++ b/src/models/decider/native/src/engine.rs @@ -1,5 +1,8 @@ //! Assemble verified artifacts, model-owned processing, and serial eager execution. -use crate::{Config, Kind, Limits, Processor, checkpoint::Checkpoint, executor::Executor}; +use crate::{ + Config, Kind, Limits, Processor, batching::BatchLimits, checkpoint::Checkpoint, + executor::Executor, +}; use anyhow::Result; use omni_runtime::SerialScheduler; use serde_json::{Value, json}; @@ -13,8 +16,15 @@ pub struct Engine { } impl Engine { pub async fn load(dir: &Path, library: &Path) -> Result { + Self::load_with_batch(dir, library, BatchLimits::from_env()?).await + } + pub async fn load_with_batch( + dir: &Path, + library: &Path, + batching: BatchLimits, + ) -> Result { let directory = dir.to_owned(); - let (processor,checkpoint,metadata) = tokio::task::spawn_blocking(move || -> Result<_> { + let (processor,checkpoint,mut metadata) = tokio::task::spawn_blocking(move || -> Result<_> { let processor = Processor::load(&directory,Limits::default())?; let labels: Vec = processor.labels().iter().map(|label| label.id).collect(); let checkpoint = Checkpoint::load(&directory,&labels)?; @@ -23,7 +33,9 @@ impl Engine { Ok((processor,checkpoint,metadata)) }).await??; let labels = processor.labels().iter().map(|label| label.id).collect(); - let executor = Executor::load(dir, library, checkpoint, labels).await?; + let executor = Executor::load(dir, library, checkpoint, labels, batching).await?; + metadata["batch_limits"] = + json!({"rows":batching.max_rows(),"tokens":batching.max_tokens()}); let engine = Self { processor: Arc::new(processor), executor, diff --git a/src/models/decider/native/src/executor.rs b/src/models/decider/native/src/executor.rs index 3868fe0e..bd832fce 100644 --- a/src/models/decider/native/src/executor.rs +++ b/src/models/decider/native/src/executor.rs @@ -1,6 +1,7 @@ //! Eager complete-row prefills and a selected tied-embedding CUDA readout. use crate::{ Limits, RowInput, + batching::BatchLimits, checkpoint::{Checkpoint, HIDDEN, LABELS, PADDED_LABELS, VOCAB}, }; use anyhow::{Context, Result, ensure}; @@ -45,6 +46,7 @@ struct Head { weights: DeviceBuffer, input: DeviceBuffer, output: DeviceBuffer, + capacity: usize, } impl Drop for Head { fn drop(&mut self) { @@ -53,7 +55,8 @@ impl Drop for Head { } } impl Head { - fn new(weights: &[u8]) -> Result { + fn new(weights: &[u8], capacity: usize) -> Result { + ensure!((1..=4).contains(&capacity), "invalid head batch capacity"); ensure!( weights.len() == PADDED_LABELS * HIDDEN * 2, "invalid selected head buffer" @@ -66,8 +69,9 @@ impl Head { stream, gemm, weights: DeviceBuffer::new(weights.len())?, - input: DeviceBuffer::new(HIDDEN * 2)?, - output: DeviceBuffer::new(PADDED_LABELS * 2)?, + input: DeviceBuffer::new(capacity * HIDDEN * 2)?, + output: DeviceBuffer::new(capacity * PADDED_LABELS * 2)?, + capacity, }; // SAFETY: the device allocation has exactly weights.len() bytes. unsafe { @@ -76,18 +80,27 @@ impl Head { Ok(head) } fn project(&self, hidden: &[f32], count: usize) -> Result> { + Ok(self.project_batch(&[hidden], &[count])?.pop().unwrap()) + } + fn project_batch(&self, hidden: &[&[f32]], counts: &[usize]) -> Result>> { + let rows = hidden.len(); ensure!( - hidden.len() == HIDDEN - && hidden.iter().all(|x| x.is_finite()) - && (2..=LABELS).contains(&count), - "invalid hidden state/readout" + rows > 0 + && rows <= self.capacity + && rows == counts.len() + && hidden + .iter() + .all(|h| h.len() == HIDDEN && h.iter().all(|x| x.is_finite())) + && counts.iter().all(|n| (2..=LABELS).contains(n)), + "invalid hidden batch/readout" ); let input: Vec = hidden .iter() + .flat_map(|h| h.iter()) .flat_map(|&x| bf16::from_f32(x).to_le_bytes()) .collect(); - // SAFETY: [1,HIDDEN] * [256,HIDDEN]^T -> [1,256]; allocated buffers match. - // The last weight row is zero padding and is never returned or normalized. + // SAFETY: [rows,HIDDEN] * [256,HIDDEN]^T -> [rows,256]; rows <= capacity. + // The final zero weight row is padding and never participates in normalization. unsafe { cuda::upload(self.input.at(0), &input, self.stream.0)?; cuda::check( @@ -96,7 +109,7 @@ impl Head { self.input.at(0), self.weights.at(0), self.output.at(0), - 1, + rows as i32, PADDED_LABELS as i32, HIDDEN as i32, PADDED_LABELS as i32, @@ -105,36 +118,60 @@ impl Head { "Decider label projection", )?; } - let mut output = vec![0; PADDED_LABELS * 2]; - // SAFETY: download waits for GEMM and copies exactly the output allocation. + let mut output = vec![0; rows * PADDED_LABELS * 2]; + // SAFETY: every returned row is inside the allocation; download synchronizes GEMM. unsafe { cuda::download(&mut output, self.output.at(0), self.stream.0)?; } - let logits: Vec = output.as_chunks::<2>().0[..count] + output + .as_chunks::<{ PADDED_LABELS * 2 }>() + .0 .iter() - .map(|&b| bf16::from_le_bytes(b).to_f32()) - .collect(); - ensure!( - logits.iter().all(|x| x.is_finite()), - "nonfinite candidate logits" - ); - Ok(logits) + .zip(counts) + .map(|(row, &count)| { + let logits: Vec = row.as_chunks::<2>().0[..count] + .iter() + .map(|&b| bf16::from_le_bytes(b).to_f32()) + .collect(); + ensure!( + logits.iter().all(|x| x.is_finite()), + "nonfinite candidate logits" + ); + Ok(logits) + }) + .collect() } } struct Loaded { model: Model, head: Head, + batching: BatchLimits, } impl Loaded { fn execute(&mut self, rows: &[RowInput]) -> Result>> { cuda::set_device(0)?; let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - rows.iter() - .map(|row| { - let hidden = self.model.forward(&row.ids)?; - self.head.project(&hidden, row.candidate_ids.len()) - }) - .collect::>>() + let lengths: Vec = rows.iter().map(|row| row.ids.len()).collect(); + let mut logits = Vec::with_capacity(rows.len()); + for range in self.batching.ranges(&lengths) { + let batch = &rows[range]; + if batch.len() == 1 { + let hidden = self.model.forward(&batch[0].ids)?; + logits.push(self.head.project(&hidden, batch[0].candidate_ids.len())?); + } else { + let inputs: Vec<&[u32]> = batch.iter().map(|row| row.ids.as_slice()).collect(); + let hidden = self.model.forward_batch(&inputs)?; + ensure!( + hidden.len() == batch.len(), + "backbone batch output count mismatch" + ); + let hidden: Vec<&[f32]> = hidden.iter().map(Vec::as_slice).collect(); + let counts: Vec = + batch.iter().map(|row| row.candidate_ids.len()).collect(); + logits.extend(self.head.project_batch(&hidden, &counts)?); + } + } + Ok(logits) })) .map_err(|_| anyhow::anyhow!("Decider execution panicked")) .and_then(|result| result); @@ -158,6 +195,7 @@ impl Executor { library: &Path, checkpoint: Checkpoint, labels: Vec, + batching: BatchLimits, ) -> Result { ensure!( std::env::var("CUA_S1_GRAPH").as_deref() != Ok("1"), @@ -166,8 +204,12 @@ impl Executor { let (dir, library) = (dir.to_owned(), library.to_owned()); let loaded = tokio::task::spawn_blocking(move || -> Result { let model = Model::load(&dir, &library)?; - let head = Head::new(&checkpoint.head)?; - Ok(Loaded { model, head }) + let head = Head::new(&checkpoint.head, batching.max_rows())?; + Ok(Loaded { + model, + head, + batching, + }) }) .await??; Ok(Self { diff --git a/src/models/decider/native/src/lib.rs b/src/models/decider/native/src/lib.rs index d51f1455..52000fe8 100644 --- a/src/models/decider/native/src/lib.rs +++ b/src/models/decider/native/src/lib.rs @@ -31,6 +31,7 @@ pub const MODEL_ID: &str = "decider-2b-v11"; pub const RUNTIME_REVISION: &str = "50d0be0d7cb43d2066965ce5fa7f3fe4e489a60f"; pub const CHECKPOINT_REVISION: &str = "533964dae8be954c5b5e19fa4948e48408094c1e"; +pub mod batching; mod checkpoint; pub mod engine; pub mod executor; diff --git a/tests/decider/batching.rs b/tests/decider/batching.rs new file mode 100644 index 00000000..becc198b --- /dev/null +++ b/tests/decider/batching.rs @@ -0,0 +1,37 @@ +use omni_decider_native::batching::BatchLimits; +#[test] +fn bounded_contiguous_ranges_preserve_order_and_long_rows() { + let limits = BatchLimits::new(3, 8).unwrap(); + assert_eq!( + limits.ranges(&[2, 3, 3, 1, 9, 2, 2, 2, 2]), + vec![0..3, 3..4, 4..5, 5..8, 8..9] + ); + assert!(limits.ranges(&[]).is_empty()); + assert_eq!(limits.ranges(&[9, 9]), vec![0..1, 1..2]); +} +#[test] +fn row_limit_and_exact_token_boundary_are_independent() { + assert_eq!( + BatchLimits::new(2, 4096).unwrap().ranges(&[1, 1, 1]), + vec![0..2, 2..3] + ); + assert_eq!( + BatchLimits::new(4, 4096).unwrap().ranges(&[2048, 2048, 1]), + vec![0..2, 2..3] + ); + assert_eq!( + BatchLimits::default().ranges(&[1, 2, 3]), + vec![0..1, 1..2, 2..3] + ); +} +#[test] +fn startup_limits_reject_unbounded_or_malformed_settings() { + for (rows, tokens) in [(0, 1), (5, 1), (8, 1), (16, 1), (1, 0), (1, 4097)] { + assert!(BatchLimits::new(rows, tokens).is_err()); + } + assert!(BatchLimits::from_values(Some("wrong"), None).is_err()); + assert!(BatchLimits::from_values(None, Some("-1")).is_err()); + let limits = BatchLimits::from_values(Some("4"), Some("4096")).unwrap(); + assert_eq!(limits.max_rows(), 4); + assert_eq!(limits.max_tokens(), 4096); +} diff --git a/tests/decider/executor.rs b/tests/decider/executor.rs index d5116dce..c5a73532 100644 --- a/tests/decider/executor.rs +++ b/tests/decider/executor.rs @@ -44,7 +44,7 @@ fn bf16_head_rounds_before_fp32_calibration() { ] { weights[offset..offset + 2].copy_from_slice(&bf16::from_f32(value).to_le_bytes()); } - let head = Head::new(&weights).unwrap(); + let head = Head::new(&weights, 1).unwrap(); let mut hidden = vec![0.0; HIDDEN]; hidden[..2].copy_from_slice(&[1.0, 1.0]); assert_eq!(head.project(&hidden, 2).unwrap(), vec![1.0, 1.015625]); @@ -52,3 +52,36 @@ fn bf16_head_rounds_before_fp32_calibration() { hidden[0] = f32::NAN; assert!(head.project(&hidden, 2).is_err()); } + +#[test] +#[ignore = "requires the built Qwen CUDA library and an available NVIDIA GPU"] +fn batched_head_preserves_rows_counts_and_bf16_rounding() { + cuda::load(Path::new(&std::env::var("DECIDER_CUDA_LIB").unwrap())).unwrap(); + cuda::set_device(0).unwrap(); + let mut weights = vec![0; PADDED_LABELS * HIDDEN * 2]; + for (offset, value) in [ + (0, 1.0), + (2, 0.00390625), + (HIDDEN * 2, 1.0), + (HIDDEN * 2 + 2, 0.015625), + ] { + weights[offset..offset + 2].copy_from_slice(&bf16::from_f32(value).to_le_bytes()); + } + let head = Head::new(&weights, 2).unwrap(); + let mut a = vec![0.0; HIDDEN]; + a[..2].copy_from_slice(&[1.0, 1.0]); + let mut b = vec![0.0; HIDDEN]; + b[..2].copy_from_slice(&[2.0, 1.0]); + let first = head.project_batch(&[&a, &b], &[2, 255]).unwrap(); + assert_eq!(first[0], vec![1.0, 1.015625]); + assert_eq!(&first[1][..2], &[2.0, 2.015625]); + assert_eq!(first[1].len(), 255); + assert!(first[1][2..].iter().all(|&x| x == 0.0)); + let reversed = head.project_batch(&[&b, &a], &[255, 2]).unwrap(); + assert_eq!(reversed[0], first[1]); + assert_eq!(reversed[1], first[0]); + assert!(head.project_batch(&[&a, &b, &a], &[2, 2, 2]).is_err()); + assert!(head.project_batch(&[&a, &b], &[2]).is_err()); + b[0] = f32::NAN; + assert!(head.project_batch(&[&a, &b], &[2, 2]).is_err()); +} diff --git a/tests/decider/fail_head.c b/tests/decider/fail_head.c index 73c9fb87..07418ff3 100644 --- a/tests/decider/fail_head.c +++ b/tests/decider/fail_head.c @@ -10,8 +10,8 @@ int cs1_gemm(void* handle, const void* x, const void* w, void* y, static unsigned projection_calls; gemm_fn original = (gemm_fn)dlsym(RTLD_NEXT, "cs1_gemm"); if (!original) return 999; - if (m == 1 && n == 256 && k == 2048) { - fprintf(stderr, "Decider test projection %u\n", ++projection_calls); + if (m >= 1 && m <= 16 && n == 256 && k == 2048) { + fprintf(stderr, "Decider test projection %u rows %d\n", ++projection_calls, m); if (projection_calls == 2) return 2; } return original(handle, x, w, y, m, n, k, ldy, stream); diff --git a/tests/decider/gpu.rs b/tests/decider/gpu.rs index 65532b8e..8bec730e 100644 --- a/tests/decider/gpu.rs +++ b/tests/decider/gpu.rs @@ -91,3 +91,47 @@ async fn maximum_complete_row_grows_scratch_without_leaking_state() { assert_eq!(engine.predict(short).await.unwrap(), baseline); assert!(engine.executor.is_ready()); } + +#[tokio::test] +#[ignore = "requires pinned Decider-2B checkpoint and NVIDIA CUDA; request packing"] +async fn packed_mixed_rows_preserve_order_usage_and_recovery() { + use omni_decider_native::batching::BatchLimits; + use serde_json::json; + let dir = std::env::var("DECIDER_MODEL").unwrap(); + let library = std::env::var("DECIDER_CUDA_LIB").unwrap(); + let engine = Engine::load_with_batch( + Path::new(&dir), + Path::new(&library), + BatchLimits::new(4, 4096).unwrap(), + ) + .await + .unwrap(); + let raw=json!({"state":"A customer requests a refund for a damaged item.","questions":{"first":{"instructions":"Choose the team.","criteria":["refund","shipping"]},"score":{"type":"score","instructions":"Severity.","criteria":["low","medium","high"]},"last":{"type":"noul","instructions":"The item is damaged."}}}).to_string(); + let first = engine.predict(raw.as_bytes()).await.unwrap(); + assert_eq!( + first["answers"] + .as_object() + .unwrap() + .keys() + .map(String::as_str) + .collect::>(), + vec!["first", "score", "last"] + ); + assert_eq!(engine.health()["batch_limits"]["rows"], 4); + let long=json!({"state":"The item is damaged. ".repeat(1000),"questions":{"q":{"instructions":"Choose the team.","criteria":["refund","shipping"]}}}).to_string(); + assert!( + engine.processor.prepare(long.as_bytes()).unwrap().rows[0] + .ids + .len() + > 4096 + ); + assert!(engine.predict(long.as_bytes()).await.is_ok()); + assert_eq!(engine.predict(raw.as_bytes()).await.unwrap(), first); + assert!( + engine + .predict(br#"{"state":0,"questions":{"q":{"criteria":["only"]}}}"#) + .await + .is_err() + ); + assert_eq!(engine.predict(raw.as_bytes()).await.unwrap(), first); +} diff --git a/tests/decider/test_modes.py b/tests/decider/test_modes.py new file mode 100644 index 00000000..d58b27f6 --- /dev/null +++ b/tests/decider/test_modes.py @@ -0,0 +1,45 @@ +import importlib.util +import json +from pathlib import Path +import sys +import tempfile +import unittest +from unittest import mock + +with mock.patch.object(sys, "path", [str(Path(__file__).parent), *sys.path]): + spec = importlib.util.spec_from_file_location("decider_modes", Path(__file__).with_name("verify_modes.py")) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + +class ModeProtocol(unittest.TestCase): + def test_revised_comparison_method_is_frozen_before_mode_launch(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + parity = root / "parity" + parity.mkdir() + (parity / "protocol.json").write_text(json.dumps({"cases": []})) + for name in ("reference.jsonl", "native.jsonl"): + (parity / name).write_text("") + binary = root / "binary" + binary.write_bytes(b"test artifact, never executed") + plan = root / "plan.json" + plan.write_text(json.dumps({"repetitions": 2, "parity_output": str(parity), "timed_cases": [], + "library": str(binary), "model": "model", "modes": [{"name": "control", "binary": str(binary), "env": {}}]})) + output = root / "output" + + def launch(*args, **kwargs): + protocol = json.loads((output / "protocol.json").read_text()) + self.assertEqual(protocol["validation_protocol"]["version"], 2) + self.assertEqual(protocol["gates"], protocol["validation_protocol"]["original_gates"]) + self.assertEqual(protocol["baseline_protocol"], {"cases": []}) + raise RuntimeError("stopped before execution") + + with mock.patch("sys.argv", ["verify_modes.py", "--plan", str(plan), "--output", str(output)]), \ + mock.patch.object(module.subprocess, "Popen", side_effect=launch): + with self.assertRaisesRegex(RuntimeError, "stopped before execution"): + module.main() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/decider/test_retirement.py b/tests/decider/test_retirement.py new file mode 100644 index 00000000..30e2399a --- /dev/null +++ b/tests/decider/test_retirement.py @@ -0,0 +1,80 @@ +import contextlib +import importlib.util +import io +import json +import os +from pathlib import Path +import tempfile +import unittest +from unittest import mock + +spec = importlib.util.spec_from_file_location( + "decider_retirement", Path(__file__).with_name("verify_retirement.py") +) +module = importlib.util.module_from_spec(spec) +spec.loader.exec_module(module) + + +class WorkerLaunch(unittest.TestCase): + def run_cli(self, batched, inherited): + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) + args = ["verify_retirement.py", "--worker", "worker", "--model", "model", + "--wrapper", "wrapper.so", "--output", str(output)] + if batched: + args.append("--batched") + + def launch(*args, **kwargs): + # Simulate external GPU evidence so this host-only test can + # inspect the actual CLI's worker launch and request boundary. + rows = 2 if batched else 1 + (output / "worker.log").write_text( + f"Decider test projection 1 rows 1\n" + f"Decider test projection 2 rows {rows}\n" + ) + return child + + child = mock.Mock(pid=123) + replies = [(503, b'{"detail":"model inference failed"}'), + (503, b'{"status":"unavailable"}'), + (503, b'{"detail":"model unavailable"}'), + (503, b'{"detail":"model unavailable"}')] + with mock.patch.dict(os.environ, inherited, clear=True), \ + mock.patch.object(module.subprocess, "Popen", side_effect=launch) as popen, \ + mock.patch.object(module, "wait_ready", return_value={"status": "ready"}), \ + mock.patch.object(module, "memory", side_effect=[4000, 0]), \ + mock.patch.object(module, "request", side_effect=replies) as request, \ + mock.patch("sys.argv", args), contextlib.redirect_stdout(io.StringIO()): + module.main() + child.terminate.assert_called_once() + child.wait.assert_called_once_with(timeout=30) + raw = json.loads(request.call_args_list[0].args[2]) + self.assertEqual(raw["questions"]["q"].get("type", "choice"), + "score" if batched else "choice") + self.assertEqual(len(raw["questions"]["q"]["criteria"]), 2) + return popen.call_args.kwargs["env"] + + def assert_mode(self, environment, rows): + self.assertEqual(environment["DECIDER_BATCH_MAX_ROWS"], str(rows)) + self.assertEqual(environment["DECIDER_BATCH_MAX_TOKENS"], "4096") + for name in ("CUA_S1_GRAPH", "DECIDER_GRAPH", "DECIDER_PREFIX", "DECIDER_FIXED"): + self.assertEqual(environment[name], "0", name) + + def test_batched_launch_without_exported_limits(self): + self.assert_mode(self.run_cli(True, {}), 2) + + def test_batched_launch_overrides_inherited_limits_and_modes(self): + inherited = {"DECIDER_BATCH_MAX_ROWS": "1", "DECIDER_BATCH_MAX_TOKENS": "1", + "CUA_S1_GRAPH": "1", "DECIDER_GRAPH": "1", + "DECIDER_PREFIX": "auto", "DECIDER_FIXED": "1"} + self.assert_mode(self.run_cli(True, inherited), 2) + + def test_unbatched_launch_overrides_inherited_limits_and_modes(self): + inherited = {"DECIDER_BATCH_MAX_ROWS": "4", "DECIDER_BATCH_MAX_TOKENS": "1", + "CUA_S1_GRAPH": "1", "DECIDER_GRAPH": "1", + "DECIDER_PREFIX": "1", "DECIDER_FIXED": "1"} + self.assert_mode(self.run_cli(False, inherited), 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/decider/verify_http.py b/tests/decider/verify_http.py index 3fcef853..595f6c64 100644 --- a/tests/decider/verify_http.py +++ b/tests/decider/verify_http.py @@ -44,6 +44,8 @@ def main(): parser.add_argument("--library", type=Path, required=True) parser.add_argument("--parity-output", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--execution", default="eager") + parser.add_argument("--native-records", type=Path, help="verify_modes records.jsonl for the selected mode") args = parser.parse_args() args.output.mkdir(parents=True, exist_ok=True) worker_base, frontend_base = "http://127.0.0.1:18110", "http://127.0.0.1:18111" @@ -57,7 +59,7 @@ def main(): processes.append(worker) metadata = wait_ready(worker_base, worker, path) assert metadata["checkpoint_revision"] == "533964dae8be954c5b5e19fa4948e48408094c1e" - assert metadata["execution"] == "eager" and metadata["dtype"] == "bfloat16" + assert metadata["execution"] == args.execution and metadata["dtype"] == "bfloat16" environment.update(OMNI_JEV_BIND="127.0.0.1:18111", OMNI_JEV_BACKEND_URL=worker_base) path = args.output / "frontend.log" logs.append(path.open("w")) @@ -66,6 +68,10 @@ def main(): wait_ready(frontend_base, frontend, path) protocol = json.loads((args.parity_output / "protocol.json").read_text()) native_records = [json.loads(line) for line in (args.parity_output / "native.jsonl").read_text().splitlines()] + if args.native_records: + native_records = [item["record"] for line in args.native_records.read_text().splitlines() + for item in [json.loads(line)] if item["phase"] == "parity"] + assert len(native_records) == len(protocol["cases"]) for case, native in zip(protocol["cases"], native_records): body = json.dumps(case["request"], ensure_ascii=False).encode() direct, proxy = request(worker_base, "/v1/systemone", body), request(frontend_base, "/v1/systemone", body) diff --git a/tests/decider/verify_modes.py b/tests/decider/verify_modes.py new file mode 100644 index 00000000..f488d0d9 --- /dev/null +++ b/tests/decider/verify_modes.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +"""Sequential matched diagnostic-runner comparisons; raw output and frozen plan retained.""" +import argparse +import hashlib +import json +import os +from pathlib import Path +import statistics +import subprocess +import time + +from verify_reference import GATES, VALIDATION_PROTOCOL, compare + + +def digest(path): + return hashlib.sha256(Path(path).read_bytes()).hexdigest() + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--plan", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + plan = json.loads(args.plan.read_text()) + assert plan["repetitions"] == 2 + parity = Path(plan["parity_output"]) + protocol = json.loads((parity / "protocol.json").read_text()) + reference = [json.loads(line) for line in (parity / "reference.jsonl").read_text().splitlines()] + baseline = [json.loads(line) for line in (parity / "native.jsonl").read_text().splitlines()] + cases = protocol["cases"] + assert len(cases) == len(reference) == len(baseline) + selected = [i for i, case in enumerate(cases) if case["name"] in plan["timed_cases"]] + assert len(selected) == len(plan["timed_cases"]) + args.output.mkdir(parents=True, exist_ok=False) + frozen = {**plan, "gates": GATES, "validation_protocol": VALIDATION_PROTOCOL, "baseline_protocol": protocol, + "input_hashes": {name: digest(parity / name) for name in ("protocol.json", "reference.jsonl", "native.jsonl")}, + "runner_sha256": digest(__file__), "comparison_sha256": digest(Path(__file__).with_name("verify_reference.py")), "library_sha256": digest(plan["library"]), + "binary_hashes": {mode["name"]: digest(mode["binary"]) for mode in plan["modes"]}, + "timing": "wall roundtrip: request JSON encoding/write through diagnostic stdout JSON decode; includes CPU preparation, GPU synchronization and serialization; excludes startup and one workload warmup; concurrency=1"} + (args.output / "protocol.json").write_text(json.dumps(frozen, indent=2) + "\n") + summaries = [] + for mode in plan["modes"]: + directory = args.output / mode["name"] + directory.mkdir() + env = dict(os.environ) + for key in ("CUA_S1_GRAPH", "DECIDER_GRAPH", "DECIDER_PREFIX", "DECIDER_FIXED", "DECIDER_BATCH_MAX_ROWS", "DECIDER_BATCH_MAX_TOKENS"): + env.pop(key, None) + env.update(mode["env"]) + with (directory / "stderr.log").open("w") as stderr, (directory / "records.jsonl").open("w") as records: + start = time.perf_counter() + child = subprocess.Popen([mode["binary"], plan["model"], plan["library"]], env=env, + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=stderr, text=True) + def run(request): + begun = time.perf_counter() + child.stdin.write(json.dumps(request, ensure_ascii=False) + "\n") + child.stdin.flush() + line = child.stdout.readline() + if not line: + raise RuntimeError(f"{mode['name']} terminated: {child.poll()}; see stderr.log") + value = json.loads(line) + return value, (time.perf_counter() - begun) * 1000 + try: + run({"state": "", "questions": {}}) + startup_ms = (time.perf_counter() - start) * 1000 + checks, samples = [], [] + for i, case in enumerate(cases): + got, ms = run(case["request"]) + records.write(json.dumps({"phase": "parity", "name": case["name"], "elapsed_ms": ms, "record": got}, ensure_ascii=False) + "\n") + records.flush() + assert got["rows"] == baseline[i]["rows"], (mode["name"], case["name"], "prepared rows") + checks.append({"name": case["name"], **compare(reference[i]["response"], got["response"])}) + for i in selected: + got, ms = run(cases[i]["request"]) + records.write(json.dumps({"phase": "warmup", "name": cases[i]["name"], "elapsed_ms": ms, "record": got}, ensure_ascii=False) + "\n") + for repetition in range(plan["repetitions"]): + for i in selected: + got, ms = run(cases[i]["request"]) + sample = {"name": cases[i]["name"], "repetition": repetition, "elapsed_ms": ms} + samples.append(sample) + records.write(json.dumps({"phase": "measured", **sample, "record": got}, ensure_ascii=False) + "\n") + records.flush() + assert got["rows"] == baseline[i]["rows"], (mode["name"], cases[i]["name"], "measured rows") + compare(reference[i]["response"], got["response"]) + memory = subprocess.check_output(["nvidia-smi", "--query-gpu=memory.used", "--format=csv,noheader,nounits"], text=True).strip() + summary = {"name": mode["name"], "startup_to_empty_diagnostic_ms": startup_ms, + "checks": checks, "samples": samples, "device_used_mib_at_end": int(memory), + "warm_ms": {name: {"median": statistics.median(values), "min": min(values), "max": max(values)} + for name in plan["timed_cases"] + for values in [[s["elapsed_ms"] for s in samples if s["name"] == name]]}} + summaries.append(summary) + (directory / "summary.json").write_text(json.dumps(summary, indent=2) + "\n") + print(json.dumps(summary), flush=True) + finally: + child.stdin.close() + try: + child.wait(timeout=30) + except subprocess.TimeoutExpired: + child.kill() + child.wait() + (args.output / "summary.json").write_text(json.dumps({"modes": summaries}, indent=2) + "\n") + + +if __name__ == "__main__": + main() diff --git a/tests/decider/verify_retirement.py b/tests/decider/verify_retirement.py index 84357b55..ca80bed6 100644 --- a/tests/decider/verify_retirement.py +++ b/tests/decider/verify_retirement.py @@ -24,11 +24,16 @@ def main(): parser.add_argument("--model", type=Path, required=True) parser.add_argument("--wrapper", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--batched", action="store_true") args = parser.parse_args() args.output.mkdir(parents=True, exist_ok=True) log = args.output / "worker.log" environment = os.environ.copy() - environment.update(DECIDER_MODEL=str(args.model), DECIDER_CUDA_LIB=str(args.wrapper), DECIDER_PORT="18112", CUA_S1_GRAPH="0") + environment.update( + DECIDER_MODEL=str(args.model), DECIDER_CUDA_LIB=str(args.wrapper), DECIDER_PORT="18112", + DECIDER_BATCH_MAX_ROWS="2" if args.batched else "1", DECIDER_BATCH_MAX_TOKENS="4096", + CUA_S1_GRAPH="0", DECIDER_GRAPH="0", DECIDER_PREFIX="0", DECIDER_FIXED="0", + ) with log.open("w") as output: child = subprocess.Popen([str(args.worker)], env=environment, stdout=output, stderr=subprocess.STDOUT) try: @@ -37,6 +42,8 @@ def main(): assert health["status"] == "ready" before = memory(child.pid) raw = b'{"state":"The item is damaged.","questions":{"q":{"instructions":"Choose a category.","criteria":["refund","shipping"]}}}' + if args.batched: + raw = b'{"state":"The item is damaged.","questions":{"q":{"type":"score","instructions":"Severity.","criteria":["low","high"]}}}' status, body = request(base, "/v1/systemone", raw) assert status == 503 and json.loads(body)["detail"] == "model inference failed" status, body = request(base, "/health") @@ -50,6 +57,8 @@ def main(): text = log.read_text() assert text.count("Decider test projection") == 2 assert "Decider test projection 2" in text + if args.batched: + assert "Decider test projection 2 rows 2" in text result = {"passed": True, "projection_calls": 2, "health_after": "unavailable", "response_status": 503, "sampled_before_mib": before, "sampled_after_mib": after, "later_requests_refused": 2} (args.output / "summary.json").write_text(json.dumps(result, indent=2) + "\n")