decider: bound pending requests and enforce HTTP deadlines

docs/architecture.md

@@ -20,11 +20,14 @@ uses a separate Hopper CUDA backend for one complete padded request. These worke
 coordinate independent processors and executors through
 `prepare` → `execute` → `finish` and use the
 [native runtime](../src/runtime/README.md) for FIFO admission and blocking dispatch
-per loaded executor. Shared processing orchestration, batch budgets,
-compatibility grouping and dynamic batching are planned.
+per loaded executor. The [Decider native worker](../recipe/decider/README.md)
+also uses this runtime and can opt into outstanding-request and processed-token
+budgets plus HTTP request deadlines. Shared processing orchestration,
+cross-request batch assembly, compatibility grouping and dynamic batching are planned.
 
 Cua-S1 and Open-Jev compute their decision heads on the CPU after downloading
-the final hidden state. JEMM uses the shared Qwen executor with separate unmerged
+the final hidden state. Decider projects its selected LM-head rows on CUDA.
+JEMM uses the shared Qwen executor with separate unmerged
 FP32 adapters and framework-reference arithmetic. It computes selected BF16
 label-head logits on CUDA, then calibrates typed answers on the CPU. JEMM admits
 one complete text or 1–4-image request and synchronizes both model streams before
@@ -154,7 +157,14 @@ executor, with FIFO waiting. Engines own scheduler instances; model-specific
 execution adapters declare the unit and submit owned closures. Waiting is async,
 so queued requests do not occupy blocking threads waiting for a model lock.
 Schedulers are local to each worker/executor; this does not coordinate separate
-processes. Queue-length limits and token budgets are not implemented yet.
+processes. Workers can separately opt into the runtime's `BoundedAdmission` for
+outstanding-request and caller-defined cost budgets. Decider's HTTP adapter
+reserves requests before upload/preparation and complete prepared-row tokens
+before execution, then carries those reservations in `ExecutionContext` through
+blocking work. The worker selects configuration and the processor defines token
+cost; shared runtime primitives own capacity accounting, cancellation/deadline
+checks and FIFO execution admission. These request budgets do not add
+cross-request batching or replace the scheduler's single execution permit.
 
 - Batch only work accepted by the same loaded executor, with compatible
   checkpoint/adapter identity, device, dtype, and input layout. Models declare

recipe/README.md

@@ -16,14 +16,16 @@ For a first real decision, follow the [complete CPU walkthrough](../docs/getting
 - [Cua-S1 4B 0.2 native text worker](cua_s1/native.md): build the CUDA library and
   the Rust worker, export BF16 weights with the original FP32 adapter and start
   the worker.
+- [Decider-2B v11 native text worker](decider/README.md): load verified pinned
+  weights, build the Rust/CUDA worker and configure optional batching, Graph,
+  prefix reuse, request/token capacity and HTTP deadlines;
+  [validation](decider/validation.md).
 - [Cua-S1 native screenshot worker](cua_s1/native_screenshot.md): export the
   multimodal language checkpoint and serve one-image decisions with Rust/CUDA.
 - [Open-Jev native text worker](open_jev/native.md): export the merged text
   backbone and trained decision head of Open-Jev-27B-v1.1 or Open-Jev-9B, then
   serve with Rust and CUDA. [Open-Jev-9B validation](open_jev/validation-9b.md)
   compares the 9B worker with the reference.
-- [Decider-2B v11 native worker](decider/README.md): pinned BF16 text decisions,
-  optional request-local batching, Graph and prefix reuse; [validation](decider/validation.md).
 - [JEMM native worker](jemm/README.md): pinned text/image decisions and export;
   known multi-image numerical limitation is documented in the model contract.
 - [JEV-27B-VL experimental worker](jev_vl/README.md): native text and preencoded

recipe/decider/README.md

@@ -36,6 +36,9 @@ Worker terminal:
 
 ```sh
 DECIDER_GRAPH=0 \
+DECIDER_MAX_PENDING_REQUESTS=8 \
+DECIDER_MAX_PENDING_TOKENS=1048576 \
+DECIDER_REQUEST_TIMEOUT_MS=5000 \
 DECIDER_MODEL=/path/to/models/decider-2b-v11 \
 DECIDER_CUDA_LIB="$PWD/target/release/libqwen3_5_cuda.so" \
 DECIDER_HOST=127.0.0.1 DECIDER_PORT=8000 \
@@ -85,9 +88,9 @@ with empty answers and zero input/output usage. No partial answers are returned.
 Supported inputs are plain text/JSON state, independent Choice (2–255 options),
 Noul and isolated Score (2–10 levels). State-prefix truncation is 32,768 tokens;
 complete rows can include question suffixes up to 36,864 tokens. At most 1,024
-expanded rows and 1,048,576 processed tokens are accepted per request. These bounds
-do not bound all waiting requests together; runtime pending-queue limits and
-cross-request batching remain separate work.
+expanded rows and 1,048,576 processed tokens are accepted per request. These processor bounds
+are separate from the optional pending-request, processed-token and deadline
+controls described below. Cross-request batching remains unsupported.
 
 The pinned worker requires released calibration. Images, video, chat/schema-first,
 packed questions, cross-request prefix caching and quantized/CPU/Metal execution
@@ -166,7 +169,7 @@ worker lifetime and keeps the completed eager result; health then reports `eager
 Launch/inference failures still retire the worker. Capture and startup costs must
 be measured separately from warm replay.
 
-## Optional request-local shared prefixes (pending integration)
+## Optional request-local shared prefixes
 
 This integration depends on the pending Qwen continuation/fixed-GEMM/executor
 stack (#98/#99). Build this integrated branch with ABI9; older libraries are
@@ -218,3 +221,85 @@ DECIDER_MODEL=/path/to/models/decider-2b-v11 \
 DECIDER_CUDA_LIB="$PWD/target/release/libqwen3_5_cuda.so" \
   target/release/omni-decider
 ```
+
+## Capacity and deadline semantics
+
+These optional HTTP limits apply to eager, packed, Graph, fixed, shared and auto
+execution. They do not apply to startup warmup or direct `Engine::predict` calls.
+The example values in the startup command are independent configuration examples,
+not measured deployment recommendations. Omit a value to leave that limit unset;
+explicit values must be positive integers and invalid values fail before loading.
+Graph remains incompatible with fixed/shared/auto; these limits do not change
+that mode constraint.
+
+| Variable | What it bounds |
+| --- | --- |
+| `DECIDER_MAX_PENDING_REQUESTS` | Outstanding HTTP requests, including body upload, preparation, execution queueing, execution and response reconstruction. A slot is reserved before reading the body. |
+| `DECIDER_MAX_PENDING_TOKENS` | The sum of complete prepared-row token lengths across admitted requests. Reservation follows preprocessing; repeated state and isolated Score rows count in full. This differs from API `usage.input_tokens` and does not change it. |
+| `DECIDER_REQUEST_TIMEOUT_MS` | One absolute deadline from request admission through response reconstruction, including slow body upload and waiting. It does not cover startup, `/health` or network delivery of the response. |
+
+Request/token capacity uses shared
+[`BoundedAdmission`](../../src/runtime/README.md#optional-request-capacity-and-execution-control).
+The separate FIFO scheduler still admits one complete Decider execution request
+at a time. Request capacity also bounds preprocessing; a token-only deployment
+must prepare a request before it can determine its token cost.
+
+Temporary request/token contention returns `503`, `Content-Type: application/json`,
+`Retry-After: 1`, and `{"detail":"inference capacity exhausted"}`. Use bounded
+retries with backoff and an overall client deadline for this response.
+
+A single request whose processed-token cost exceeds the entire token ceiling
+returns JSON `422` with `{"detail":"request exceeds processed-token capacity"}`
+and no `Retry-After` header. It is rejected before execution and cannot succeed
+unchanged under that configuration: reduce its prepared work or raise the
+configured budget and restart the worker. A request costing exactly the ceiling
+is accepted when that token capacity is free.
+
+Deadline expiry returns JSON `504` with `{"detail":"request timed out"}`.
+The deadline is strict: work that completes after it may still yield `504`
+instead of a successful response. Timeout/disconnect signals cancellation and
+prevents work that has not started from running at the next context check;
+it does not preempt already-running CPU/CUDA work. Detached preparation and
+execution retain their capacity until they finish, and submitted device work
+must synchronize before its resources are released. A `504` therefore does
+not imply an immediately free request slot.
+
+Existing content-type, payload and validation checks remain in force: JSON is
+required (`415` otherwise), the request body limit is 8 MiB (`413`), and processor
+validation errors return `422`. A not-ready worker returns `503` with
+`{"detail":"model unavailable"}` rather than the capacity-refusal response.
+
+Within-request packing is configured separately with `DECIDER_BATCH_MAX_ROWS`
+(default `1`, allowed `1..=4`) and `DECIDER_BATCH_MAX_TOKENS` (default `4096`,
+allowed `1..=4096`). These select packed groups within an admitted request;
+they do not replace the outstanding-token budget. Rows above the packing token
+budget run alone, subject to the processor/executor's separate input limits.
+
+## Health and validation
+
+`GET /health` is outside request admission and returns `200` for a ready
+executor or `503` for an unavailable one. Existing model, checkpoint, execution,
+batch and calibration metadata remain present. When any HTTP runtime limit is
+set, `runtime_limits` reports `requests`, `processed_tokens` and `timeout_ms`;
+unset members are `null`.
+
+Only a configured request-count bound adds top-level `depth`, `capacity` and
+`rejected`. `depth` includes all outstanding request reservations, including
+blocking work retained after a timeout; `capacity` is the request-count ceiling.
+`rejected` counts request-capacity and token-capacity refusals, including requests
+larger than the entire token ceiling. It excludes timeouts and processor
+validation failures. Token-only or timeout-only deployments do not publish
+these three fields. Token occupancy is not exposed by this health interface.
+
+The configuration and HTTP control checks use no checkpoint or GPU:
+
+```sh
+cargo test --locked -p omni-runtime
+cargo test --locked -p omni-decider-native --test runtime
+cargo test --locked -p omni-decider-native --lib serve::runtime_tests
+```
+
+These checks cover admission, cancellation, capacity retention and synthetic
+HTTP deadlines. They do not establish model parity, performance, GPU-memory
+limits or other-hardware support. Device/checkpoint qualification remains a
+separate check with the pinned artifacts and matching rebuilt CUDA library.

src/models/decider/native/Cargo.toml

@@ -16,7 +16,7 @@ safetensors = "0.8.0"
 omni-qwen3-5-native = { path = "../../qwen3_5/native" }
 omni-runtime = { path = "../../../runtime" }
 axum = "0.8.8"
-tokio = { version = "1.49.0", features = ["macros", "net", "rt-multi-thread", "sync", "signal"] }
+tokio = { version = "1.49.0", features = ["macros", "net", "rt-multi-thread", "sync", "signal", "time"] }
 
 [[bin]]
 name = "omni-decider"
@@ -49,3 +49,14 @@ path = "../../../../tests/decider/prefix.rs"
 [[example]]
 name = "decider-run"
 path = "../../../../tests/decider/runner.rs"
+
+[[test]]
+name = "runtime"
+path = "../../../../tests/decider/runtime.rs"
+
+[[test]]
+name = "startup"
+path = "../../../../tests/decider/startup.rs"
+
+[dev-dependencies]
+tokio = { version = "1.49.0", features = ["io-util"] }

src/models/decider/native/src/engine.rs

@@ -1,7 +1,10 @@
 //! Assemble verified artifacts, model-owned processing, and serial eager execution.
 use crate::{
-    Config, Kind, Limits, Processor, batching::BatchLimits, checkpoint::Checkpoint,
+    Config, Kind, Limits, Processor,
+    batching::BatchLimits,
+    checkpoint::Checkpoint,
     executor::Executor,
+    runtime::{Runtime, RuntimeConfig},
 };
 use anyhow::Result;
 use omni_runtime::SerialScheduler;
@@ -12,17 +15,19 @@ pub struct Engine {
     pub processor: Arc<Processor>,
     pub executor: Executor,
     pub scheduler: SerialScheduler,
+    pub runtime: Runtime,
     metadata: Value,
 }
 impl Engine {
     pub async fn load(dir: &Path, library: &Path) -> Result<Self> {
         let graph = crate::options::graph_env()?;
-        Self::load_with_modes(
+        Self::load_with_modes_and_runtime(
             dir,
             library,
             BatchLimits::from_env()?,
             graph,
             crate::options::prefix_env(graph)?,
+            RuntimeConfig::from_env()?,
         )
         .await
     }
@@ -31,7 +36,7 @@ impl Engine {
         library: &Path,
         batching: BatchLimits,
     ) -> Result<Self> {
-        Self::load_with_options(dir, library, batching, false).await
+        Self::load_with_runtime(dir, library, batching, RuntimeConfig::from_env()?).await
     }
     pub async fn load_with_options(
         dir: &Path,
@@ -54,11 +59,48 @@ impl Engine {
         batching: BatchLimits,
         graph: bool,
         prefix_mode: crate::prefix::PrefixMode,
+    ) -> Result<Self> {
+        Self::load_with_modes_and_runtime(
+            dir,
+            library,
+            batching,
+            graph,
+            prefix_mode,
+            RuntimeConfig::from_env()?,
+        )
+        .await
+    }
+    /// Explicit runtime configuration with the existing independent eager defaults.
+    pub async fn load_with_runtime(
+        dir: &Path,
+        library: &Path,
+        batching: BatchLimits,
+        config: RuntimeConfig,
+    ) -> Result<Self> {
+        Self::load_with_modes_and_runtime(
+            dir,
+            library,
+            batching,
+            false,
+            crate::prefix::PrefixMode::Off,
+            config,
+        )
+        .await
+    }
+    /// Select execution and HTTP policy together without reading environment switches.
+    pub async fn load_with_modes_and_runtime(
+        dir: &Path,
+        library: &Path,
+        batching: BatchLimits,
+        graph: bool,
+        prefix_mode: crate::prefix::PrefixMode,
+        config: RuntimeConfig,
     ) -> Result<Self> {
         anyhow::ensure!(
             !graph || prefix_mode == crate::prefix::PrefixMode::Off,
             "shared/fixed execution requires Graph off"
         );
+        let runtime = Runtime::new(config)?;
         let directory = dir.to_owned();
         let (processor,checkpoint,mut metadata) = tokio::task::spawn_blocking(move || -> Result<_> {
             let processor = Processor::load(&directory,Limits::default())?;
@@ -86,6 +128,7 @@ impl Engine {
             processor: Arc::new(processor),
             executor,
             scheduler: SerialScheduler::default(),
+            runtime,
             metadata,
         };
         // Loading returns only after a real complete model decision, before any socket binds.
@@ -120,6 +163,16 @@ impl Engine {
         } else {
             "unavailable"
         });
+        let config = self.runtime.config();
+        if config.max_requests.is_some() || config.max_tokens.is_some() || config.timeout.is_some()
+        {
+            metadata["runtime_limits"] = json!({"requests":config.max_requests,"processed_tokens":config.max_tokens,"timeout_ms":config.timeout.map(|timeout|timeout.as_millis())});
+        }
+        if let Some(report) = self.runtime.report() {
+            metadata["depth"] = json!(report.depth);
+            metadata["capacity"] = json!(report.capacity);
+            metadata["rejected"] = json!(report.rejected);
+        }
         metadata
     }
 }

src/models/decider/native/src/executor.rs

@@ -12,6 +12,7 @@ use omni_qwen3_5_native::{
     model::{GraphStats, Model},
 };
 use omni_runtime::SerialScheduler;
+use omni_runtime::control::ExecutionContext;
 use std::{
     ffi::c_void,
     path::Path,
@@ -293,6 +294,16 @@ impl Executor {
         scheduler: &SerialScheduler,
         rows: Vec<RowInput>,
     ) -> Result<Vec<Vec<f32>>> {
+        self.execute_with_context(scheduler, rows, ExecutionContext::default())
+            .await
+    }
+    pub async fn execute_with_context(
+        &self,
+        scheduler: &SerialScheduler,
+        rows: Vec<RowInput>,
+        context: ExecutionContext,
+    ) -> Result<Vec<Vec<f32>>> {
+        context.check()?;
         validate_rows(&rows, &self.labels)?;
         ensure!(self.is_ready(), "executor unavailable");
         if rows.is_empty() {
@@ -305,7 +316,7 @@ impl Executor {
         );
         let prefix_stats = self.prefix_stats.clone();
         scheduler
-            .run(move || {
+            .run_with_context(context, move || {
                 let mut guard = match loaded.lock() {
                     Ok(guard) => guard,
                     Err(_) => {

src/models/decider/native/src/lib.rs

@@ -35,6 +35,7 @@ pub mod batching;
 mod checkpoint;
 pub mod engine;
 pub mod executor;
+pub mod runtime;
 pub mod serve;
 
 pub mod options;