CAGRA Batched Search And Microbatching Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Deliver ordered explicit vector and embedding batch search plus transparent CAGRA single-query microbatching, with owner-local distributed execution and an A4000-proven cuda-oxide batch path.

Architecture: Add owned homogeneous batch adapters at the Nebuchadnezzar API boundary, then carry the batch through CAGRA service, portable RPC DTOs, and the in-process distributed coordinator. Morpheus routes CAGRA requests through one bounded worker thread per VectorIndexer, preserving cuda-oxide thread-local resident/module/workspace caches; CPU and active-delta work remain ordered per query while base-segment CUDA work uses the existing one-CTA-per-query batch kernel. The persistent CAGRA format and runtime ownership boundary do not change.

Tech Stack: Rust nightly nightly-2026-04-03, Nebuchadnezzar BoxFuture APIs, Morpheus CAGRA, async-channel, Tokio oneshot channels, serde/bincode RPC DTOs, cuda-oxide, CUDA 13.x tools, RTX A4000 (sm_86)

Execution Boundaries

The approved behavior is defined in CAGRA_BATCHED_SEARCH_DESIGN.md in the Morpheus repository root.
Morpheus implementation work happens in /home/shisoft/Code/OSS Projects/cagra-engine-worktrees/Morpheus-cuda-oxide on cagra-cuda-oxide-spike.
Nebuchadnezzar API work happens in /home/shisoft/Code/OSS Projects/cagra-engine-worktrees/Nebuchadnezzar.
The Morpheus worktree already contains accepted CUDA optimization edits in src/apps/cagra/gpu/cuda_oxide/search.rs and src/apps/cagra/gpu/tests.rs. Preserve them and verify them before adding batch work.
The Morpheus worktree also contains unrelated user edits. Every commit command below stages exact paths; never use git add -A, git add ., a hard reset, or a checkout that discards local changes.
CAGRA storage artifacts, manifests, row locators, checksums, and segment layouts remain unchanged.
CAGRA buffers remain owned by cuda-oxide. Sparse cudarc buffers are not shared, copied, or imported into this path.
Only the local RTX A4000 is a required CUDA verification target. Compile kernels for sm_86.
Before Tasks 9-11, use the cuda-skill workflow for cuda-oxide API checks, Nsight counter selection, and Compute Sanitizer interpretation. Re-read the CAGRA paper (arXiv:2308.15136) as the traversal reference; batching may add independent query parallelism but must not weaken candidate expansion, termination, or deterministic top-k semantics.
Explicit batches are non-empty, homogeneous, ordered, and all-or-nothing. Transparent microbatches retry failed members individually.
This plan adds portable batch DTOs and exercises the existing in-process distributed coordinator. It does not invent a new network transport.

File Structure

Nebuchadnezzar

src/index/vector/mod.rs: vector core/coordinator batch trait methods, sequential defaults, and client forwarding.
src/index/embedding/mod.rs: embedding batch trait method, sequential default, and client forwarding.

Morpheus

src/apps/cagra/types.rs: homogeneous local batch contracts and hashable search configuration.
src/apps/cagra/rpc.rs: portable batch request/response DTOs and checked wire conversion.
src/apps/cagra/service.rs: manifest-pinned local batch execution and CPU/CUDA base-segment routing.
src/apps/cagra/coordinator.rs: owner batch fan-out, strict response validation, and per-query deterministic merge.
src/apps/cagra/batch.rs: bounded dedicated-thread transparent microbatcher and its isolated executor seam.
src/apps/cagra/mod.rs: exports for the new contracts and batcher.
src/apps/cagra/gpu/config.rs: batching defaults and validation.
src/apps/hnsw/mod.rs: VectorIndexer explicit batch overrides and CAGRA-only microbatch routing.
src/apps/embedding/mod.rs: one model resolution, one provider batch inference, one vector batch call, and ordered document mapping.
src/apps/cagra/gpu/cuda_oxide/search.rs: reusable pinned query staging and the benchmark-gated shared-query experiment.
src/apps/cagra/gpu/cuda_common.rs: query-cache shared-memory resource admission if the experiment is retained.
src/apps/cagra/tests.rs: service, RPC, local-owner, and distributed batch contracts. This file is already dirty; append narrowly and stage it only with deliberate diff review.
src/apps/cagra/gpu/tests.rs: CUDA batch equivalence, source ownership checks, and sanitizer fixtures.
src/apps/embedding/tests.rs: explicit embedding batch and CAGRA routing tests. This file is already dirty; append narrowly and review the staged diff.
src/config/tests.rs: human-readable batching configuration tests.
config/server.yaml: documented production defaults.
src/bin/cagra_cuda_oxide_embedding_e2e.rs: explicit text batch through embedding and CAGRA on CUDA.
src/bin/cagra_microbatch_profile.rs: reproducible direct-single, explicit-batch, and transparent-microbatch A4000 measurements.

Acceptance Gates

Every batch response has exactly one ordered result list per query.
CPU batch output equals repeated CPU single searches, including base plus active delta.
Distributed batch output equals repeated distributed single searches and rejects generation, coverage, or response-count violations.
Concurrent compatible singles form one executor batch; incompatible keys never mix.
A failed transparent combined batch retries each live request independently.
Explicit embedding batch calls embed_queries_batch once and reaches vector batch search once.
CUDA service batch equals CPU and repeated CUDA singles; memcheck, racecheck, and synccheck are clean.
Resident A4000 batch-64 median is no slower than 0.597 ms/query, which is the approved 5% ceiling over 0.569 ms/query.
Compatible concurrent singles through the microbatcher deliver at least 1.5x serialized direct-single throughput.
Pinned staging regresses no tested resident dimension by more than 3%.
Shared query caching is retained only when the hard resident profile improves by at least 5%, no tested dimension regresses by more than 3%, and achieved occupancy does not decline enough to erase throughput.

Task 0: Preserve And Seal The Existing CUDA Baseline

Files:

Verify: src/apps/cagra/gpu/cuda_oxide/search.rs
Verify: src/apps/cagra/gpu/tests.rs
Step 1: Record both repository states before editing

Run in the Morpheus implementation worktree:

git status --short --branch
git diff -- src/apps/cagra/gpu/cuda_oxide/search.rs src/apps/cagra/gpu/tests.rs

Run in the Nebuchadnezzar worktree:

git status --short --branch

Expected: the known Morpheus CUDA optimization diff is visible, unrelated dirty files remain unstaged, and Nebuchadnezzar is clean.

Step 2: Verify the accepted CUDA baseline on the A4000

cargo +nightly-2026-04-03 test --features cagra-cuda --lib apps::cagra::gpu::tests::cuda::production_cuda_search_source_exposes_batch_entrypoints
cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_cuda_oxide_batch_search_smoke
CAGRA_CUDA_PROFILE_ROWS=4096 CAGRA_CUDA_PROFILE_DIM=64 CAGRA_CUDA_PROFILE_DEGREE=32 CAGRA_CUDA_PROFILE_K=10 CAGRA_CUDA_PROFILE_SEARCH_WIDTH=32 CAGRA_CUDA_PROFILE_MAX_ITERATIONS=64 CAGRA_CUDA_PROFILE_QUERIES=64 CAGRA_CUDA_PROFILE_BATCH_SIZE=64 CAGRA_CUDA_PROFILE_WARMUP=5 CAGRA_CUDA_PROFILE_ITERS=20 cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_cuda_oxide_profile

Expected: the correctness test passes and the profile emits a resident batch-64 median close to the recorded 0.569 ms/query baseline.

Step 3: Seal only the accepted baseline files when they are still uncommitted

git diff --check -- src/apps/cagra/gpu/cuda_oxide/search.rs src/apps/cagra/gpu/tests.rs
git add src/apps/cagra/gpu/cuda_oxide/search.rs src/apps/cagra/gpu/tests.rs
git diff --cached --name-only
git commit -m "perf(cagra): optimize cuda-oxide batch traversal"

Expected: the staged-name output contains exactly the two CUDA files. If both files are already clean at execution time, record the current HEAD and skip the empty commit.

Task 1: Add Nebuchadnezzar Vector Batch Contracts

Files:

Modify: src/index/vector/mod.rs
Test: src/index/vector/mod.rs
Step 1: Write failing default-adapter and client-forwarding tests

Extend the existing test RecordingVectorCore so search copies the first query value into a hit, then add:

#[test]
fn vector_batch_default_adapter_preserves_query_order() {
    futures::executor::block_on(async {
        let calls = Arc::new(Mutex::new(Vec::new()));
        let core = RecordingVectorCore::new(calls);
        let results = core
            .search_batch(
                7,
                9,
                vec![vec![3.0, 0.0], vec![1.0, 0.0], vec![2.0, 0.0]],
                1,
                Some(32),
            )
            .await
            .expect("batch adapter should succeed");

        assert_eq!(
            results
                .iter()
                .map(|hits| hits[0].score)
                .collect::<Vec<_>>(),
            vec![3.0, 1.0, 2.0]
        );
    });
}

#[test]
fn vector_batch_default_adapter_rejects_empty_input() {
    futures::executor::block_on(async {
        let core = RecordingVectorCore::new(Arc::new(Mutex::new(Vec::new())));
        let error = core
            .search_batch(7, 9, Vec::new(), 1, Some(32))
            .await
            .expect_err("empty vector batch must fail");
        assert!(error.to_string().contains("must not be empty"));
    });
}

Implement RecordingVectorCore::search for these tests with owned values before entering the future:

fn search(
    &self,
    _schema_id: u32,
    _field_id: u64,
    query_vector: &[f32],
    _limit: usize,
    _ef_search: Option<u16>,
) -> BoxFuture<'_, Result<Vec<VectorHit>, IndexError>> {
    let score = query_vector[0];
    Box::pin(async move {
        Ok(vec![VectorHit {
            id: Id::new(0, score.to_bits() as u64),
            score,
        }])
    })
}

Step 2: Run the focused test and observe the missing method

cargo test vector_batch_default_adapter --lib

Expected: compilation fails because VectorIndexerCore::search_batch does not exist.

Step 3: Add owned ordered batch methods to both traits

Add this default to VectorIndexerCore:

fn search_batch(
    &self,
    schema_id: u32,
    field_id: u64,
    query_vectors: Vec<Vec<f32>>,
    limit: usize,
    ef_search: Option<u16>,
) -> BoxFuture<'_, Result<Vec<Vec<VectorHit>>, IndexError>> {
    Box::pin(async move {
        if query_vectors.is_empty() {
            return Err(IndexError::Other(
                "vector search batch must not be empty".to_string(),
            ));
        }
        let mut results = Vec::with_capacity(query_vectors.len());
        for query in query_vectors {
            results.push(
                self.search(schema_id, field_id, query.as_slice(), limit, ef_search)
                    .await?,
            );
        }
        Ok(results)
    })
}

Add the same ordered adapter to VectorSearchCoordinator, calling search_distributed, and name it search_distributed_batch. Add matching VectorIndexClient::search_batch and VectorIndexClient::search_distributed_batch forwarding methods that take Vec<Vec<f32>> by value.

Step 4: Verify the vector contract

rustfmt --edition 2021 --config skip_children=true src/index/vector/mod.rs
cargo test vector_batch_default_adapter --lib
cargo test index::vector::tests --lib

Expected: all vector module tests pass.

Step 5: Commit in Nebuchadnezzar

git add src/index/vector/mod.rs
git commit -m "feat(vector): add ordered batch search API"

Task 2: Add And Validate CAGRA Batching Configuration

Files:

Modify: src/apps/cagra/gpu/config.rs
Modify: src/config/tests.rs
Modify: config/server.yaml
Step 1: Write failing default, parse, and validation tests

Add these assertions to src/config/tests.rs:

assert!(config.cagra_gpu.batching.enabled);
assert_eq!(config.cagra_gpu.batching.max_batch_size, 64);
assert_eq!(config.cagra_gpu.batching.max_delay_us, 200);
assert_eq!(config.cagra_gpu.batching.queue_capacity, 4096);

In the explicit YAML fixture add:

  batching:
    enabled: false
    max_batch_size: 16
    max_delay_us: 75
    queue_capacity: 128

and assert those exact values. Add focused unit tests in gpu/config.rs:

#[test]
fn batching_config_rejects_zero_batch_size() {
    let config = CagraSearchBatchingConfig {
        max_batch_size: 0,
        ..CagraSearchBatchingConfig::default()
    };
    assert_eq!(
        config.validate().unwrap_err(),
        "CAGRA batching max_batch_size must be greater than zero"
    );
}

#[test]
fn batching_config_requires_queue_to_hold_one_batch() {
    let config = CagraSearchBatchingConfig {
        max_batch_size: 64,
        queue_capacity: 32,
        ..CagraSearchBatchingConfig::default()
    };
    assert!(config.validate().unwrap_err().contains("queue_capacity"));
}

Step 2: Run tests and observe missing configuration fields

cargo +nightly-2026-04-03 test --lib config::tests::test_cagra_gpu_config
cargo +nightly-2026-04-03 test --lib apps::cagra::gpu::config::tests

Expected: compilation fails because batching and CagraSearchBatchingConfig are absent.

Step 3: Implement one shared serde configuration type

Add to src/apps/cagra/gpu/config.rs:

const DEFAULT_CAGRA_SEARCH_BATCH_SIZE: usize = 64;
const DEFAULT_CAGRA_SEARCH_BATCH_DELAY_US: u64 = 200;
const DEFAULT_CAGRA_SEARCH_QUEUE_CAPACITY: usize = 4096;

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct CagraSearchBatchingConfig {
    pub enabled: bool,
    pub max_batch_size: usize,
    pub max_delay_us: u64,
    pub queue_capacity: usize,
}

impl Default for CagraSearchBatchingConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            max_batch_size: DEFAULT_CAGRA_SEARCH_BATCH_SIZE,
            max_delay_us: DEFAULT_CAGRA_SEARCH_BATCH_DELAY_US,
            queue_capacity: DEFAULT_CAGRA_SEARCH_QUEUE_CAPACITY,
        }
    }
}

impl CagraSearchBatchingConfig {
    pub fn validate(&self) -> Result<(), String> {
        if self.max_batch_size == 0 {
            return Err("CAGRA batching max_batch_size must be greater than zero".to_string());
        }
        if self.queue_capacity < self.max_batch_size {
            return Err(format!(
                "CAGRA batching queue_capacity {} must be at least max_batch_size {}",
                self.queue_capacity, self.max_batch_size
            ));
        }
        Ok(())
    }
}

Add pub batching: CagraSearchBatchingConfig to both GPU configuration structs, initialize it in both defaults, and copy it in From<HumanReadableCagraGpuConfig>.

Step 4: Add production defaults and verify parsing

Add to config/server.yaml beneath cagra_gpu:

  batching:
    enabled: true
    max_batch_size: 64
    max_delay_us: 200
    queue_capacity: 4096

Run:

rustfmt +nightly-2026-04-03 --edition 2021 src/apps/cagra/gpu/config.rs src/config/tests.rs
cargo +nightly-2026-04-03 test --lib config::tests::test_cagra_gpu_config
cargo +nightly-2026-04-03 test --lib apps::cagra::gpu::config::tests

Expected: explicit and default configuration tests pass, including zero delay as a valid value.

Step 5: Commit configuration only

git add src/apps/cagra/gpu/config.rs src/config/tests.rs config/server.yaml
git commit -m "feat(cagra): configure search batching"

Task 3: Add Portable Local And RPC Batch Contracts

Files:

Modify: src/apps/cagra/types.rs
Modify: src/apps/cagra/rpc.rs
Modify: src/apps/cagra/mod.rs
Test: src/apps/cagra/tests.rs
Step 1: Write failing contract and wire-roundtrip tests

Append tests using values above u32::MAX for every wire-sized search field:

#[test]
fn cagra_batch_rpc_request_round_trips_without_narrowing() {
    let request = CagraBatchSearchRpcRequest {
        schema_id: 10,
        field_id: 20,
        manifest_generation: u32::MAX as u64 + 9,
        segment_ids: vec![u32::MAX as u64 + 11],
        queries: vec![vec![1.0, 2.0], vec![3.0, 4.0]],
        k: 7,
        search_width: 31,
        max_iterations: 257,
    };
    let bytes = bincode::serde::encode_to_vec(&request, bincode::config::standard()).unwrap();
    let (decoded, consumed): (CagraBatchSearchRpcRequest, usize) =
        bincode::serde::decode_from_slice(&bytes, bincode::config::standard()).unwrap();
    assert_eq!(consumed, bytes.len());
    assert_eq!(decoded, request);

    let local = CagraLocalBatchSearchRequest::try_from(decoded).unwrap();
    assert_eq!(local.queries.len(), 2);
    assert_eq!(local.config.max_iterations, 257);
}

#[test]
fn cagra_batch_rpc_response_preserves_query_boundaries() {
    let response = CagraBatchSearchRpcResponse {
        manifest_generation: 17,
        owner_server_id: 23,
        searched_segment_ids: vec![29, 31],
        hits_by_query: vec![
            vec![CagraSearchRpcHit::from(CagraSearchHit {
                cell_id: Id::new(1, 2),
                distance: 0.25,
            })],
            Vec::new(),
        ],
    };
    assert_eq!(response.hits_by_query.len(), 2);
    assert_eq!(response.hits_by_query[0][0].distance, 0.25);
}

Step 2: Run and observe missing batch DTOs

cargo +nightly-2026-04-03 test --lib cagra_batch_rpc

Expected: compilation fails on the missing batch types.

Step 3: Add local contracts and make the grouping configuration hashable

Change CagraSearchConfig to derive Eq, Hash in addition to its current derives, then add:

#[derive(Debug, Clone, PartialEq)]
pub struct CagraLocalBatchSearchRequest {
    pub schema_id: u32,
    pub field_id: u64,
    pub manifest_generation: u64,
    pub segment_ids: Vec<u64>,
    pub queries: Vec<Vec<f32>>,
    pub config: CagraSearchConfig,
}

#[derive(Debug, Clone, PartialEq)]
pub struct CagraLocalBatchSearchResponse {
    pub manifest_generation: u64,
    pub owner_server_id: u64,
    pub searched_segment_ids: Vec<u64>,
    pub hits_by_query: Vec<Vec<CagraSearchHit>>,
}

Step 4: Add checked portable wire types

Add to rpc.rs:

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CagraBatchSearchRpcRequest {
    pub schema_id: u32,
    pub field_id: u64,
    pub manifest_generation: u64,
    pub segment_ids: Vec<u64>,
    pub queries: Vec<Vec<f32>>,
    pub k: u64,
    pub search_width: u64,
    pub max_iterations: u64,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CagraBatchSearchRpcResponse {
    pub manifest_generation: u64,
    pub owner_server_id: u64,
    pub searched_segment_ids: Vec<u64>,
    pub hits_by_query: Vec<Vec<CagraSearchRpcHit>>,
}

Implement checked conversion with the existing usize_from_wire helper:

impl TryFrom<CagraBatchSearchRpcRequest> for CagraLocalBatchSearchRequest {
    type Error = CagraRpcConversionError;

    fn try_from(value: CagraBatchSearchRpcRequest) -> Result<Self, Self::Error> {
        Ok(Self {
            schema_id: value.schema_id,
            field_id: value.field_id,
            manifest_generation: value.manifest_generation,
            segment_ids: value.segment_ids,
            queries: value.queries,
            config: CagraSearchConfig {
                k: usize_from_wire("k", value.k)?,
                search_width: usize_from_wire("search_width", value.search_width)?,
                max_iterations: usize_from_wire("max_iterations", value.max_iterations)?,
            },
        })
    }
}

impl From<CagraLocalBatchSearchResponse> for CagraBatchSearchRpcResponse {
    fn from(value: CagraLocalBatchSearchResponse) -> Self {
        Self {
            manifest_generation: value.manifest_generation,
            owner_server_id: value.owner_server_id,
            searched_segment_ids: value.searched_segment_ids,
            hits_by_query: value
                .hits_by_query
                .into_iter()
                .map(|hits| hits.into_iter().map(CagraSearchRpcHit::from).collect())
                .collect(),
        }
    }
}

Keep backend selection, GPU traces, pointers, and buffer identifiers out of both DTOs.

Step 5: Verify and commit the portable boundary

rustfmt +nightly-2026-04-03 --edition 2021 --config skip_children=true src/apps/cagra/types.rs src/apps/cagra/rpc.rs src/apps/cagra/mod.rs src/apps/cagra/tests.rs
cargo +nightly-2026-04-03 test --lib cagra_batch_rpc
git diff --check -- src/apps/cagra/types.rs src/apps/cagra/rpc.rs src/apps/cagra/mod.rs src/apps/cagra/tests.rs
git add src/apps/cagra/types.rs src/apps/cagra/rpc.rs src/apps/cagra/mod.rs src/apps/cagra/tests.rs
git diff --cached --name-only
git commit -m "feat(cagra): add portable batch search contracts"

Expected: both tests pass; only the four named files are staged.

Task 4: Implement Manifest-Pinned Service Batch Search

Files:

Modify: src/apps/cagra/service.rs
Test: src/apps/cagra/tests.rs
Test: src/apps/cagra/gpu/tests.rs
Modify: src/bin/cagra_cuda_oxide_batch_search_smoke.rs
Step 1: Write CPU batch parity and validation tests

Use a service with three active-delta vectors, then compare one batch with repeated singles:

#[test]
fn cagra_service_batch_matches_repeated_cpu_searches() {
    let service = CagraIndexService::new_for_tests();
    service
        .new_index(1, 2, MetricEncoding::L2, 2, sample_service_config())
        .unwrap();
    for (id, vector) in [
        (Id::new(0, 1), [0.0, 0.0]),
        (Id::new(0, 2), [1.0, 0.0]),
        (Id::new(0, 3), [3.0, 0.0]),
    ] {
        service.insert_vector(1, 2, id, &vector).unwrap();
    }
    let queries = vec![vec![0.1, 0.0], vec![2.9, 0.0]];
    let config = sample_delta_search_config(2);
    let batch = service.search_batch(1, 2, &queries, config).unwrap();
    let singles = queries
        .iter()
        .map(|query| service.search(1, 2, query, config).unwrap())
        .collect::<Vec<_>>();
    assert_eq!(batch, singles);
}

#[test]
fn cagra_service_batch_rejects_empty_and_mismatched_queries() {
    let service = CagraIndexService::new_for_tests();
    service
        .new_index(1, 2, MetricEncoding::L2, 2, sample_service_config())
        .unwrap();
    let config = sample_delta_search_config(1);
    assert!(service.search_batch(1, 2, &[], config).is_err());
    assert!(matches!(
        service.search_batch(1, 2, &[vec![1.0]], config),
        Err(CagraError::DimensionMismatch { expected: 2, actual: 1 })
    ));
}

#[test]
fn cagra_service_batch_deduplicates_cells_and_breaks_ties_by_id() {
    let service = CagraIndexService::new_for_tests();
    service
        .new_index(1, 2, MetricEncoding::L2, 1, sample_service_config())
        .unwrap();
    service.insert_vector(1, 2, Id::new(0, 2), &[-1.0]).unwrap();
    service.insert_vector(1, 2, Id::new(0, 1), &[1.0]).unwrap();
    service.insert_vector(1, 2, Id::new(0, 5), &[3.0]).unwrap();
    service.compact_active_delta(1, 2, 99).unwrap();
    service.remove_vector(1, 2, Id::new(0, 5)).unwrap();
    service.insert_vector(1, 2, Id::new(0, 5), &[0.0]).unwrap();

    let hits = service
        .search_batch(1, 2, &[vec![0.0]], sample_delta_search_config(3))
        .unwrap()
        .remove(0);
    assert_eq!(hits[0].cell_id, Id::new(0, 5));
    assert_eq!(hits[1].cell_id, Id::new(0, 1));
    assert_eq!(hits[2].cell_id, Id::new(0, 2));
    assert_eq!(hits.iter().filter(|hit| hit.cell_id == Id::new(0, 5)).count(), 1);
}

Step 2: Run and observe the missing service API

cargo +nightly-2026-04-03 test --lib cagra_service_batch

Expected: compilation fails because CagraIndexService::search_batch is absent.

Step 3: Add public and owner-local batch entry points

Add these methods:

pub fn search_batch(
    &self,
    schema_id: u32,
    field_id: u64,
    queries: &[Vec<f32>],
    config: CagraSearchConfig,
) -> Result<Vec<Vec<CagraSearchHit>>, CagraError> {
    let (_, hits_by_query) = self.search_index_batch_with_scope(
        schema_id,
        field_id,
        queries,
        config,
        None,
        None,
    )?;
    Ok(hits_by_query)
}

pub fn search_local_batch(
    &self,
    request: CagraLocalBatchSearchRequest,
) -> Result<CagraLocalBatchSearchResponse, CagraError> {
    let (generation, hits_by_query) = self.search_index_batch_with_scope(
        request.schema_id,
        request.field_id,
        request.queries.as_slice(),
        request.config,
        Some(request.manifest_generation),
        Some(request.segment_ids.as_slice()),
    )?;
    Ok(CagraLocalBatchSearchResponse {
        manifest_generation: generation,
        owner_server_id: self.local_server_id,
        searched_segment_ids: request.segment_ids,
        hits_by_query,
    })
}

Step 4: Implement one-snapshot batch execution

Implement search_index_batch_with_scope by reproducing the existing scope validation exactly once, then allocate vec![Vec::new(); queries.len()]. Reject an empty batch before index lookup. Validate every query against the pinned manifest dimension before launching any backend. Search active delta per query. Load each selected base segment once, call search_base_segment_with_backend_batch, append results by query index, then filter and merge each result independently:

for (query_index, hits) in hits_by_query.iter_mut().enumerate() {
    let live = self.filter_live_hits_in_manifest(&manifest, std::mem::take(hits))?;
    *hits = super::merge::merge_cagra_hits(live, config.k);
    debug_assert!(query_index < queries.len());
}

For an empty registered index return (0, vec![Vec::new(); queries.len()]), while preserving current generation and segment-scope error text.

Step 5: Route base segments through CPU or ordered CUDA chunks

Add search_base_segment_with_backend_batch with the same route admission as the single method. The CPU branch loops in input order. The CUDA branch splits queries by self.cagra_gpu_config.batching.max_batch_size and invokes the existing facade:

self.cagra_gpu_config
    .batching
    .validate()
    .map_err(CagraError::InvalidConfig)?;
let mut results = Vec::with_capacity(queries.len());
for chunk in queries.chunks(self.cagra_gpu_config.batching.max_batch_size) {
    results.extend(
        super::gpu::cuda::search::search_base_segment_graph_cuda_batch_with_budgets(
            manifest.schema_id,
            manifest.field_id,
            descriptor,
            segment,
            chunk,
            config,
            self.cagra_gpu_config.preferred_device_ordinal,
            self.cagra_gpu_config.device_memory_budget_bytes,
            self.cagra_gpu_config.cache_capacity_bytes,
        )?,
    );
}
if results.len() != queries.len() {
    return Err(CagraError::Storage(format!(
        "CAGRA base batch returned {} results for {} queries",
        results.len(),
        queries.len()
    )));
}
if self.cagra_gpu_config.explain_traces {
    for (query_index, result) in results.iter().enumerate() {
        if let Some(trace) = &result.gpu_trace {
            log::debug!(
                target: "morpheus::cagra::batch",
                "segment={} query_index={} query_count={} chunk_limit={} cache_hit={} vector_h2d_bytes={} adjacency_h2d_bytes={} query_h2d_bytes={} result_d2h_bytes={} h2d_seconds={:?} kernel_seconds={:?} d2h_seconds={:?} total_seconds={:?}",
                trace.segment_id,
                query_index,
                queries.len(),
                self.cagra_gpu_config.batching.max_batch_size,
                trace.cache_hit,
                trace.vector_h2d_bytes,
                trace.adjacency_h2d_bytes,
                trace.query_h2d_bytes,
                trace.result_d2h_bytes,
                trace.host_to_device_seconds,
                trace.kernel_seconds,
                trace.device_to_host_seconds,
                trace.total_seconds,
            );
        }
    }
}
Ok(results)

Keep Vulkan's existing unavailable/error route. Do not silently fall back when RequireCuda is selected.

Step 6: Add a base-plus-delta test and a RequireCuda service test

Build a base, add a new delta row, and compare the mixed batch with repeated singles:

#[test]
fn cagra_service_base_plus_delta_batch_matches_singles() {
    let service = CagraIndexService::new_for_tests();
    service
        .new_index(1, 2, MetricEncoding::L2, 2, sample_service_config())
        .unwrap();
    service.insert_vector(1, 2, Id::new(0, 1), &[0.0, 0.0]).unwrap();
    service.insert_vector(1, 2, Id::new(0, 2), &[2.0, 0.0]).unwrap();
    service.compact_active_delta(1, 2, 99).unwrap();
    service.insert_vector(1, 2, Id::new(0, 3), &[1.0, 0.0]).unwrap();

    let queries = vec![vec![0.1, 0.0], vec![1.1, 0.0]];
    let config = sample_delta_search_config(3);
    let batch = service.search_batch(1, 2, &queries, config).unwrap();
    let singles = queries
        .iter()
        .map(|query| service.search(1, 2, query, config).unwrap())
        .collect::<Vec<_>>();
    assert_eq!(batch, singles);
}

Under #[cfg(feature = "cagra-cuda")], use the same portable store for CPU and RequireCuda services:

#[test]
fn cagra_service_cuda_batch_matches_cpu_and_cuda_singles() {
    let store = InMemoryCagraStore::default();
    let cpu = CagraIndexService::new_with_store_and_gpu_config_for_tests(
        store.clone(),
        1,
        CagraGpuConfig {
            execution: CagraExecutionPreference::Cpu,
            ..CagraGpuConfig::default()
        },
    );
    cpu.new_index(1, 2, MetricEncoding::L2, 2, sample_service_config())
        .unwrap();
    for row in 0..64_u64 {
        cpu.insert_vector(1, 2, Id::new(0, row + 1), &[row as f32, 0.0])
            .unwrap();
    }
    cpu.compact_active_delta(1, 2, 99).unwrap();

    let cuda = CagraIndexService::recover_from_store_with_gpu_config_for_tests(
        store,
        1,
        CagraGpuConfig {
            execution: CagraExecutionPreference::RequireCuda,
            ..CagraGpuConfig::default()
        },
    )
    .unwrap();
    let queries = vec![vec![0.2, 0.0], vec![62.8, 0.0]];
    let config = CagraSearchConfig { k: 4, search_width: 16, max_iterations: 64 };
    let expected = queries
        .iter()
        .map(|query| cpu.search(1, 2, query, config).unwrap())
        .collect::<Vec<_>>();
    let actual = cuda.search_batch(1, 2, &queries, config).unwrap();
    let singles = queries
        .iter()
        .map(|query| cuda.search(1, 2, query, config).unwrap())
        .collect::<Vec<_>>();
    assert_eq!(actual, expected);
    assert_eq!(actual, singles);
}

Mirror the RequireCuda service assertion in cagra_cuda_oxide_batch_search_smoke.rs so device execution is compiled by cargo oxide:

fn run_service_batch_parity() {
    let service = CagraIndexService::new_with_store_and_gpu_config(
        InMemoryCagraStore::default(),
        1,
        CagraGpuConfig {
            execution: CagraExecutionPreference::RequireCuda,
            ..CagraGpuConfig::default()
        },
    );
    service
        .new_index(
            1,
            2,
            MetricEncoding::L2,
            4,
            CagraConfig {
                graph_degree: 8,
                intermediate_graph_degree: 16,
                delta_graph_degree: 8,
                max_delta_rows: 1_024,
                build_algo: CagraBuildAlgo::BruteForceKnn,
            },
        )
        .unwrap();
    for row in 0..32_u64 {
        service
            .insert_vector(1, 2, Id::new(0, row + 1), &[row as f32, 0.0, 0.0, 0.0])
            .unwrap();
    }
    service.compact_active_delta(1, 2, 99).unwrap();
    let queries = vec![vec![0.1, 0.0, 0.0, 0.0], vec![30.9, 0.0, 0.0, 0.0]];
    let config = CagraSearchConfig { k: 4, search_width: 8, max_iterations: 32 };
    let batch = service.search_batch(1, 2, &queries, config).unwrap();
    let singles = queries
        .iter()
        .map(|query| service.search(1, 2, query, config).unwrap())
        .collect::<Vec<_>>();
    assert_eq!(batch, singles);
}

The smoke's existing direct batch comparison remains the CPU graph-search oracle.

Step 7: Verify service and CUDA integration

cargo +nightly-2026-04-03 test --lib cagra_service_batch
cargo +nightly-2026-04-03 test --features cagra-cuda --lib --no-run
cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_cuda_oxide_batch_search_smoke

Expected: CPU, base-plus-delta, chunking, and RequireCuda tests pass.

Step 8: Commit service batch execution

git add src/apps/cagra/service.rs src/apps/cagra/tests.rs src/apps/cagra/gpu/tests.rs src/bin/cagra_cuda_oxide_batch_search_smoke.rs
git diff --cached --check
git commit -m "feat(cagra): execute manifest-pinned search batches"

Task 5: Add Strict Distributed Batch Coordination

Files:

Modify: src/apps/cagra/coordinator.rs
Test: src/apps/cagra/tests.rs
Step 1: Write distributed parity and malformed-owner tests

Extend the existing two-owner fixture with two queries:

#[test]
fn cagra_distributed_batch_matches_repeated_distributed_singles() {
    let first = CagraIndexService::new_with_store_for_tests(InMemoryCagraStore::default(), 1);
    let second = CagraIndexService::new_with_store_for_tests(InMemoryCagraStore::default(), 2);
    for service in [&first, &second] {
        service
            .new_index(1, 2, MetricEncoding::L2, 2, sample_service_config())
            .unwrap();
    }
    first.insert_vector(1, 2, Id::new(0, 1), &[0.0, 0.0]).unwrap();
    second.insert_vector(1, 2, Id::new(0, 2), &[10.0, 10.0]).unwrap();
    let coordinator = InProcessCagraCoordinator::new(vec![first, second]);
    let queries = vec![vec![0.0, 0.0], vec![10.0, 10.0]];
    let config = sample_delta_search_config(2);
    let batch = coordinator.search_batch(1, 2, &queries, config).unwrap();
    let singles = queries
        .iter()
        .map(|query| coordinator.search(1, 2, query, config).unwrap())
        .collect::<Vec<_>>();
    assert_eq!(batch, singles);
}

Add direct validator coverage for all malformed response dimensions:

#[test]
fn cagra_distributed_batch_rejects_malformed_owner_responses() {
    let valid = CagraLocalBatchSearchResponse {
        manifest_generation: 7,
        owner_server_id: 3,
        searched_segment_ids: vec![11, 13],
        hits_by_query: vec![Vec::new(), Vec::new()],
    };
    assert!(validate_batch_owner_response(8, &[11, 13], 2, &valid).is_err());

    let missing_segment = CagraLocalBatchSearchResponse {
        manifest_generation: 7,
        searched_segment_ids: vec![11],
        ..valid.clone()
    };
    assert!(validate_batch_owner_response(7, &[11, 13], 2, &missing_segment).is_err());

    let short_results = CagraLocalBatchSearchResponse {
        manifest_generation: 7,
        hits_by_query: vec![Vec::new()],
        ..valid
    };
    assert!(validate_batch_owner_response(7, &[11, 13], 2, &short_results).is_err());
}

Add a coordinator-level generation test:

#[test]
fn cagra_distributed_batch_rejects_mixed_manifest_generations() {
    let first = CagraIndexService::new_with_store_for_tests(InMemoryCagraStore::default(), 1);
    let second = CagraIndexService::new_with_store_for_tests(InMemoryCagraStore::default(), 2);
    for service in [&first, &second] {
        service
            .new_index(1, 2, MetricEncoding::L2, 2, sample_service_config())
            .unwrap();
    }
    first.insert_vector(1, 2, Id::new(0, 1), &[0.0, 0.0]).unwrap();
    second.insert_vector(1, 2, Id::new(0, 2), &[1.0, 0.0]).unwrap();
    second.insert_vector(1, 2, Id::new(0, 3), &[2.0, 0.0]).unwrap();

    let error = InProcessCagraCoordinator::new(vec![first, second])
        .search_batch(1, 2, &[vec![0.0, 0.0]], sample_delta_search_config(1))
        .expect_err("mixed generations must fail the whole batch");
    assert!(error.to_string().contains("distributed manifest generation mismatch"));
}

Step 2: Run and observe the missing coordinator method

cargo +nightly-2026-04-03 test --lib cagra_distributed_batch

Expected: compilation fails because search_batch is absent.

Step 3: Implement owner fan-out and response validation

Add InProcessCagraCoordinator::search_batch. Reject empty owners and empty queries. Call local_search_plan once per selected owner before sending requests, pin the first returned generation, and reject any other owner plan whose generation differs:

let mut owner_plans = Vec::with_capacity(self.owners.len());
let mut pinned_generation = None;
for owner in &self.owners {
    let (generation, segment_ids) = owner.local_search_plan(schema_id, field_id)?;
    match pinned_generation {
        Some(expected) if generation != expected => {
            return Err(CagraError::Storage(format!(
                "CAGRA distributed manifest generation mismatch: expected {expected}, found {generation}"
            )));
        }
        None => pinned_generation = Some(generation),
        _ => {}
    }
    owner_plans.push((owner, generation, segment_ids));
}

Then call search_local_batch once per planned owner and validate before merging:

pub(crate) fn validate_batch_owner_response(
    requested_generation: u64,
    requested_segments: &[u64],
    query_count: usize,
    response: &CagraLocalBatchSearchResponse,
) -> Result<(), CagraError> {
    if response.manifest_generation != requested_generation {
        return Err(CagraError::Storage(format!(
            "CAGRA owner {} returned manifest generation {}, expected {}",
            response.owner_server_id,
            response.manifest_generation,
            requested_generation
        )));
    }
    if response.hits_by_query.len() != query_count {
        return Err(CagraError::Storage(format!(
            "CAGRA owner {} returned {} query results, expected {}",
            response.owner_server_id,
            response.hits_by_query.len(),
            query_count
        )));
    }
    let mut expected = requested_segments.to_vec();
    let mut actual = response.searched_segment_ids.clone();
    expected.sort_unstable();
    actual.sort_unstable();
    if actual != expected {
        return Err(CagraError::Storage(format!(
            "CAGRA owner {} segment coverage {:?} does not match requested {:?}",
            response.owner_server_id, actual, expected
        )));
    }
    Ok(())
}

Accumulate hits by query index and call merge_cagra_hits independently after every owner response is validated. Never merge by owner arrival order.

Measure fan-out and merge separately and log query_count, owner count, covered segment count, and merge microseconds with target morpheus::cagra::distributed. The log carries total batch durations, not per-query normalized values.

Step 4: Verify strict completeness and commit

rustfmt +nightly-2026-04-03 --edition 2021 src/apps/cagra/coordinator.rs src/apps/cagra/tests.rs
cargo +nightly-2026-04-03 test --lib cagra_distributed_batch
cargo +nightly-2026-04-03 test --lib cagra_distributed_search
git add src/apps/cagra/coordinator.rs src/apps/cagra/tests.rs
git commit -m "feat(cagra): coordinate distributed search batches"

Expected: parity, malformed response, generation mismatch, and segment coverage tests pass.

Task 6: Implement The Dedicated Transparent Microbatch Worker

Files:

Create: src/apps/cagra/batch.rs
Modify: src/apps/cagra/mod.rs
Test: src/apps/cagra/batch.rs
Step 1: Write deterministic executor tests before the worker

Define a test executor that records exact keys and sizes. Its batch method rejects any empty vector while its single method rejects only that member; this makes fallback observable:

#[derive(Default)]
struct RecordingExecutor {
    batches: Mutex<Vec<(CagraBatchKey, usize)>>,
    single_calls: AtomicUsize,
}

impl RecordingExecutor {
    fn batch_sizes(&self) -> Vec<usize> {
        self.batches
            .lock()
            .iter()
            .map(|(_, size)| *size)
            .collect()
    }
}

impl CagraBatchExecutor for RecordingExecutor {
    fn search(
        &self,
        _schema_id: u32,
        _field_id: u64,
        query: &[f32],
        _config: CagraSearchConfig,
    ) -> Result<Vec<CagraSearchHit>, CagraError> {
        self.single_calls.fetch_add(1, Ordering::SeqCst);
        let Some(score) = query.first().copied() else {
            return Err(CagraError::InvalidConfig("empty fixture query".to_string()));
        };
        Ok(vec![CagraSearchHit {
            cell_id: Id::new(0, score.to_bits() as u64),
            distance: score,
        }])
    }

    fn search_batch(
        &self,
        schema_id: u32,
        field_id: u64,
        queries: &[Vec<f32>],
        config: CagraSearchConfig,
    ) -> Result<Vec<Vec<CagraSearchHit>>, CagraError> {
        self.batches.lock().push((CagraBatchKey { schema_id, field_id, config }, queries.len()));
        if queries.iter().any(Vec::is_empty) {
            return Err(CagraError::InvalidConfig("fixture batch contains empty query".to_string()));
        }
        Ok(queries
            .iter()
            .map(|query| vec![CagraSearchHit {
                cell_id: Id::new(0, query[0].to_bits() as u64),
                distance: query[0],
            }])
            .collect())
    }
}

fn test_search_config() -> CagraSearchConfig {
    CagraSearchConfig {
        k: 1,
        search_width: 4,
        max_iterations: 8,
    }
}

fn test_batcher(
    executor: Arc<RecordingExecutor>,
    max_batch_size: usize,
    max_delay_us: u64,
) -> CagraSearchBatcher {
    CagraSearchBatcher::new(
        executor,
        CagraSearchBatchingConfig {
            max_batch_size,
            max_delay_us,
            queue_capacity: max_batch_size.max(8),
            ..CagraSearchBatchingConfig::default()
        },
    )
    .unwrap()
}

Add the compatible-group test:

#[tokio::test]
async fn compatible_singles_form_one_microbatch() {
    let executor = Arc::new(RecordingExecutor::default());
    let batcher = CagraSearchBatcher::new(
        executor.clone(),
        CagraSearchBatchingConfig {
            max_batch_size: 4,
            max_delay_us: 50_000,
            queue_capacity: 8,
            ..CagraSearchBatchingConfig::default()
        },
    )
    .unwrap();
    let config = CagraSearchConfig { k: 1, search_width: 4, max_iterations: 8 };
    let futures = (0..4).map(|value| {
        batcher.search(1, 2, vec![value as f32, 0.0], config)
    });
    let results = futures::future::join_all(futures).await;
    assert!(results.iter().all(Result::is_ok));
    assert_eq!(executor.batch_sizes(), vec![4]);
}

Add exact-key, deadline, chunking, and failure-isolation tests:

#[tokio::test]
async fn incompatible_configs_never_share_a_microbatch() {
    let executor = Arc::new(RecordingExecutor::default());
    let batcher = test_batcher(executor.clone(), 8, 2_000);
    let left = CagraSearchConfig { k: 1, search_width: 4, max_iterations: 8 };
    let right = CagraSearchConfig { k: 2, search_width: 4, max_iterations: 8 };
    let (left_result, right_result) = tokio::join!(
        batcher.search(1, 2, vec![1.0], left),
        batcher.search(1, 2, vec![2.0], right),
    );
    assert!(left_result.is_ok() && right_result.is_ok());
    let batches = executor.batches.lock();
    assert_eq!(batches.len(), 2);
    assert_ne!(batches[0].0, batches[1].0);
}

#[tokio::test]
async fn deadline_flushes_a_partial_group() {
    let executor = Arc::new(RecordingExecutor::default());
    let batcher = test_batcher(executor.clone(), 8, 1_000);
    let result = tokio::time::timeout(
        Duration::from_secs(1),
        batcher.search(1, 2, vec![1.0], test_search_config()),
    )
    .await
    .expect("deadline should flush")
    .unwrap();
    assert_eq!(result[0].distance, 1.0);
    assert_eq!(executor.batches.lock()[0].1, 1);
}

#[tokio::test]
async fn maximum_batch_size_chunks_in_input_order() {
    let executor = Arc::new(RecordingExecutor::default());
    let batcher = test_batcher(executor.clone(), 2, 5_000);
    let results = futures::future::join_all((0..5).map(|value| {
        batcher.search(1, 2, vec![value as f32], test_search_config())
    }))
    .await;
    assert_eq!(
        results.into_iter().map(|result| result.unwrap()[0].distance).collect::<Vec<_>>(),
        vec![0.0, 1.0, 2.0, 3.0, 4.0]
    );
    assert_eq!(
        executor.batches.lock().iter().map(|(_, size)| *size).collect::<Vec<_>>(),
        vec![2, 2, 1]
    );
}

#[tokio::test]
async fn failed_microbatch_retries_members_individually() {
    let executor = Arc::new(RecordingExecutor::default());
    let batcher = test_batcher(executor.clone(), 2, 5_000);
    let (valid, invalid) = tokio::join!(
        batcher.search(1, 2, vec![7.0], test_search_config()),
        batcher.search(1, 2, Vec::new(), test_search_config()),
    );
    assert_eq!(valid.unwrap()[0].distance, 7.0);
    assert!(invalid.is_err());
    assert_eq!(executor.single_calls.load(Ordering::SeqCst), 2);
}

#[tokio::test]
async fn explicit_batch_failure_is_not_retried_as_singles() {
    let executor = Arc::new(RecordingExecutor::default());
    let batcher = test_batcher(executor.clone(), 8, 0);
    assert!(batcher
        .search_batch(1, 2, vec![vec![1.0], Vec::new()], test_search_config())
        .await
        .is_err());
    assert_eq!(executor.single_calls.load(Ordering::SeqCst), 0);
}

Expose a private enqueue_single helper that returns the response receiver after bounded send. Use a blocking executor to prove capacity, cancellation, and shutdown:

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn bounded_queue_skips_cancelled_work_and_worker_exits() {
    let (entered_tx, entered_rx) = std::sync::mpsc::channel();
    let (release_tx, release_rx) = std::sync::mpsc::channel();
    let executor = Arc::new(BlockingExecutor::new(entered_tx, release_rx));
    let batcher = CagraSearchBatcher::new(
        executor.clone(),
        CagraSearchBatchingConfig {
            max_batch_size: 1,
            max_delay_us: 0,
            queue_capacity: 1,
            ..CagraSearchBatchingConfig::default()
        },
    )
    .unwrap();

    let first_batcher = batcher.clone();
    let first = tokio::spawn(async move {
        first_batcher.search(1, 2, vec![1.0], test_search_config()).await
    });
    entered_rx.recv_timeout(Duration::from_secs(1)).unwrap();

    let cancelled = batcher
        .enqueue_single(1, 2, vec![2.0], test_search_config())
        .await
        .unwrap();
    assert!(tokio::time::timeout(
        Duration::from_millis(20),
        batcher.enqueue_single(1, 2, vec![3.0], test_search_config()),
    )
    .await
    .is_err());
    drop(cancelled);
    release_tx.send(()).unwrap();
    assert!(first.await.unwrap().is_ok());

    tokio::time::sleep(Duration::from_millis(20)).await;
    assert_eq!(executor.launched_query_values(), vec![vec![1.0]]);
    let metrics = batcher.metrics_for_tests();
    drop(batcher);
    tokio::time::timeout(Duration::from_secs(1), async {
        while !metrics.worker_exited.load(Ordering::Acquire) {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("dropping all senders should stop the worker");
}

Use this blocking executor:

struct BlockingExecutor {
    entered: Mutex<Option<std::sync::mpsc::Sender<()>>>,
    release: Mutex<Option<std::sync::mpsc::Receiver<()>>>,
    launched: Mutex<Vec<Vec<f32>>>,
}

impl BlockingExecutor {
    fn new(
        entered: std::sync::mpsc::Sender<()>,
        release: std::sync::mpsc::Receiver<()>,
    ) -> Self {
        Self {
            entered: Mutex::new(Some(entered)),
            release: Mutex::new(Some(release)),
            launched: Mutex::new(Vec::new()),
        }
    }

    fn launched_query_values(&self) -> Vec<Vec<f32>> {
        self.launched.lock().clone()
    }
}

fn fixture_hit(value: f32) -> CagraSearchHit {
    CagraSearchHit {
        cell_id: Id::new(0, value.to_bits() as u64),
        distance: value,
    }
}

impl CagraBatchExecutor for BlockingExecutor {
    fn search(
        &self,
        _schema_id: u32,
        _field_id: u64,
        query: &[f32],
        _config: CagraSearchConfig,
    ) -> Result<Vec<CagraSearchHit>, CagraError> {
        Ok(vec![fixture_hit(query[0])])
    }

    fn search_batch(
        &self,
        _schema_id: u32,
        _field_id: u64,
        queries: &[Vec<f32>],
        _config: CagraSearchConfig,
    ) -> Result<Vec<Vec<CagraSearchHit>>, CagraError> {
        self.launched
            .lock()
            .push(queries.iter().map(|query| query[0]).collect());
        if let Some(sender) = self.entered.lock().take() {
            sender.send(()).unwrap();
        }
        if let Some(receiver) = self.release.lock().take() {
            receiver.recv().unwrap();
        }
        Ok(queries.iter().map(|query| vec![fixture_hit(query[0])]).collect())
    }
}

Step 2: Run and observe the missing module

cargo +nightly-2026-04-03 test --lib apps::cagra::batch

Expected: compilation fails because batch.rs and CagraSearchBatcher do not exist.

Step 3: Define the executor seam and work contracts

Create batch.rs with:

pub trait CagraBatchExecutor: Send + Sync + 'static {
    fn search(
        &self,
        schema_id: u32,
        field_id: u64,
        query: &[f32],
        config: CagraSearchConfig,
    ) -> Result<Vec<CagraSearchHit>, CagraError>;

    fn search_batch(
        &self,
        schema_id: u32,
        field_id: u64,
        queries: &[Vec<f32>],
        config: CagraSearchConfig,
    ) -> Result<Vec<Vec<CagraSearchHit>>, CagraError>;
}

impl CagraBatchExecutor for CagraIndexService {
    fn search(&self, schema_id: u32, field_id: u64, query: &[f32], config: CagraSearchConfig)
        -> Result<Vec<CagraSearchHit>, CagraError>
    {
        CagraIndexService::search(self, schema_id, field_id, query, config)
    }

    fn search_batch(&self, schema_id: u32, field_id: u64, queries: &[Vec<f32>], config: CagraSearchConfig)
        -> Result<Vec<Vec<CagraSearchHit>>, CagraError>
    {
        CagraIndexService::search_batch(self, schema_id, field_id, queries, config)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
struct CagraBatchKey {
    schema_id: u32,
    field_id: u64,
    config: CagraSearchConfig,
}

Use an enum with Single and Explicit variants. Each variant owns its queries and a Tokio oneshot sender. Add shared monotonic metrics for observability and deterministic tests:

#[derive(Debug, Default)]
pub struct CagraBatchMetrics {
    batch_launches: AtomicU64,
    launched_queries: AtomicU64,
    individual_fallbacks: AtomicU64,
    full_flushes: AtomicU64,
    deadline_flushes: AtomicU64,
    worker_exited: AtomicBool,
}

#[derive(Clone)]
pub struct CagraSearchBatcher {
    sender: async_channel::Sender<CagraBatchWork>,
    config: CagraSearchBatchingConfig,
    metrics: Arc<CagraBatchMetrics>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CagraBatchMetricsSnapshot {
    pub batch_launches: u64,
    pub launched_queries: u64,
    pub individual_fallbacks: u64,
    pub full_flushes: u64,
    pub deadline_flushes: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CagraBatchFlushReason {
    Full,
    Deadline,
    Explicit,
    Shutdown,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CagraBatchObservation {
    pub batch_size: usize,
    pub queue_wait_us_by_query: Vec<u64>,
    pub reason: CagraBatchFlushReason,
    pub individual_fallbacks: usize,
}

pub type CagraBatchObserver = Arc<dyn Fn(CagraBatchObservation) + Send + Sync>;

impl CagraSearchBatcher {
    pub fn new<E>(
        executor: Arc<E>,
        config: CagraSearchBatchingConfig,
    ) -> Result<Self, CagraError>
    where
        E: CagraBatchExecutor,
    {
        Self::new_with_observer(executor, config, None)
    }

    pub fn new_with_observer<E>(
        executor: Arc<E>,
        config: CagraSearchBatchingConfig,
        observer: Option<CagraBatchObserver>,
    ) -> Result<Self, CagraError>
    where
        E: CagraBatchExecutor,
    {
        config.validate().map_err(CagraError::InvalidConfig)?;
        let executor: Arc<dyn CagraBatchExecutor> = executor;
        Self::spawn(executor, config, observer)
    }

    pub fn metrics_snapshot(&self) -> CagraBatchMetricsSnapshot {
        CagraBatchMetricsSnapshot {
            batch_launches: self.metrics.batch_launches.load(Ordering::Relaxed),
            launched_queries: self.metrics.launched_queries.load(Ordering::Relaxed),
            individual_fallbacks: self.metrics.individual_fallbacks.load(Ordering::Relaxed),
            full_flushes: self.metrics.full_flushes.load(Ordering::Relaxed),
            deadline_flushes: self.metrics.deadline_flushes.load(Ordering::Relaxed),
        }
    }

    #[cfg(test)]
    fn metrics_for_tests(&self) -> Arc<CagraBatchMetrics> {
        Arc::clone(&self.metrics)
    }
}

Expose an immutable numeric snapshot for diagnostics. new installs no observer; new_with_observer accepts an optional observer for the profile binary and tests. Build per-query queue-wait vectors only when an observer is installed. Under #[cfg(test)], return Arc<CagraBatchMetrics> from metrics_for_tests so shutdown can be observed without retaining a queue sender.

Step 4: Spawn one bounded worker thread without an ownership cycle

Validate configuration, create async_channel::bounded(queue_capacity), and move only the receiver and executor into a named thread:

let (sender, receiver) = async_channel::bounded(config.queue_capacity);
let metrics = Arc::new(CagraBatchMetrics::default());
let worker_metrics = Arc::clone(&metrics);
std::thread::Builder::new()
    .name("morpheus-cagra-search".to_string())
    .spawn(move || {
        run_batch_worker(
            receiver,
            executor,
            worker_config,
            worker_metrics,
            observer,
        )
    })
    .map_err(|error| CagraError::Storage(format!(
        "failed to start CAGRA search worker: {error}"
    )))?;
Ok(Self { sender, config, metrics })

Do not store a sender in the worker. Channel closure must flush live pending requests and terminate the thread.

Step 5: Implement bounded async submission

search and search_batch create oneshot channels, await bounded sender.send, then await the response. Map a closed worker channel to CagraError::Storage("CAGRA search worker stopped".to_string()). Explicit batch rejects empty input before enqueueing and is sent as one Explicit work item.

Step 6: Implement deadline/full grouping and individual fallback

Maintain HashMap<CagraBatchKey, PendingGroup> where each group stores first_enqueued_at and ordered requests. Poll try_recv, sleep for at most min(remaining_deadline, 25 microseconds) when no new item is ready, flush immediately at max_batch_size, and flush partial groups at their own deadline. Process explicit items on the same worker without joining them to singles.

Before launch, remove entries whose response sender is closed. On a combined error, retry every remaining request with executor.search; send each independent result to its original oneshot. On success, require results.len() == live_requests.len() or treat the mismatch as a combined failure and use the same individual fallback.

Log query_count, effective chunk size, queue wait microseconds, full or deadline, and fallback count with the morpheus::cagra::batch log target. Durations remain batch durations unless explicitly normalized.

Step 7: Run the complete worker contract suite

rustfmt +nightly-2026-04-03 --edition 2021 src/apps/cagra/batch.rs
git diff --check -- src/apps/cagra/batch.rs src/apps/cagra/mod.rs
cargo +nightly-2026-04-03 test --lib apps::cagra::batch -- --nocapture

Do not pass src/apps/cagra/mod.rs to standalone rustfmt: rustfmt recursively formats its child modules and can overwrite unrelated worktree changes. Expected: all grouping, deadline, capacity, cancellation, fallback, explicit-batch, and shutdown tests pass without a GPU.

Step 8: Commit the worker

git add src/apps/cagra/batch.rs src/apps/cagra/mod.rs
git commit -m "feat(cagra): add bounded search microbatch worker"

Task 7: Route Vector Singles And Explicit Batches Through CAGRA

Files:

Modify: src/apps/hnsw/mod.rs
Test: src/apps/hnsw/mod.rs
Step 1: Write routing tests

Test the extracted CAGRA helper without constructing an HNSW RPC service:

#[tokio::test]
async fn cagra_vector_batch_uses_one_ordered_executor_launch() {
    let service = CagraIndexService::new_for_tests();
    service
        .new_index(
            1,
            2,
            MetricEncoding::L2,
            2,
            neb::index::vector::CagraConfig {
                graph_degree: 2,
                intermediate_graph_degree: 4,
                delta_graph_degree: 2,
                max_delta_rows: 1_024,
                build_algo: neb::index::vector::CagraBuildAlgo::Auto,
            },
        )
        .unwrap();
    service.insert_vector(1, 2, Id::new(0, 1), &[0.0, 0.0]).unwrap();
    service.insert_vector(1, 2, Id::new(0, 2), &[2.0, 0.0]).unwrap();
    let batcher = CagraSearchBatcher::new(
        Arc::new(service.clone()),
        CagraSearchBatchingConfig {
            max_batch_size: 4,
            max_delay_us: 0,
            queue_capacity: 8,
            ..CagraSearchBatchingConfig::default()
        },
    )
    .unwrap();

    let results = VectorIndexer::search_cagra_batch(
        &service,
        &batcher,
        true,
        1,
        2,
        vec![vec![0.1, 0.0], vec![1.9, 0.0]],
        1,
        Some(16),
    )
    .await
    .unwrap();
    assert_eq!(results[0][0].id, Id::new(0, 1));
    assert_eq!(results[1][0].id, Id::new(0, 2));
    assert_eq!(batcher.metrics_snapshot().batch_launches, 1);
    assert_eq!(batcher.metrics_snapshot().launched_queries, 2);
}

Keep the existing HNSW tests in the focused suite; the batch override's non-CAGRA branch must call the existing single HNSW helper in input order and leave the batcher's metrics unchanged.

Step 2: Run and observe missing trait overrides

cargo +nightly-2026-04-03 test --lib apps::hnsw::tests::cagra_vector_batch

Expected: compilation fails because the Morpheus VectorIndexer has no batch overrides or batcher field.

Step 3: Construct and retain one batcher per indexer

Add cagra_batcher: CagraSearchBatcher to VectorIndexer. In new_for_runtime, construct it immediately after service recovery:

let cagra_batcher = CagraSearchBatcher::new(
    Arc::new(cagra.clone()),
    runtime.cagra_gpu_config().batching.clone(),
)
.map_err(|error| format!("failed to initialize CAGRA search batcher: {error}"))?;

When test helpers replace indexer.cagra, rebuild indexer.cagra_batcher from the replacement service so tests never dispatch to a stale service clone.

Step 4: Split CAGRA and HNSW search helpers

Keep HNSW's existing async coordinator path unchanged. For registered CAGRA indexes, use the batcher only when batching.enabled; otherwise call the service directly. Add a mapping helper:

fn map_cagra_hits(hits: Vec<CagraSearchHit>) -> Vec<VectorHit> {
    hits.into_iter()
        .map(|hit| VectorHit { id: hit.cell_id, score: hit.distance })
        .collect()
}

Extract the testable explicit helper with this signature:

async fn search_cagra_batch(
    cagra: &CagraIndexService,
    batcher: &CagraSearchBatcher,
    batching_enabled: bool,
    schema_id: u32,
    field_id: u64,
    queries: Vec<Vec<f32>>,
    limit: usize,
    ef_search: Option<u16>,
) -> Result<Vec<Vec<VectorHit>>, IndexError>

When enabled, it awaits batcher.search_batch; when disabled, it calls cagra.search_batch synchronously on the current owner. Both branches map errors through map_cagra_error and map each hit list through map_cagra_hits.

The existing single search must enqueue exactly one query without sleeping on the async runtime thread.

Step 5: Override both vector batch traits

Implement VectorIndexerCore::search_batch and VectorSearchCoordinator::search_distributed_batch. Reject empty input. If CAGRA is registered, send one explicit batch work item and map every hit list in order. Otherwise call the corresponding Neb default behavior explicitly by looping existing HNSW single calls; do not enqueue HNSW work into the CAGRA worker.

The current production CAGRA distributed baseline has no network transport. The coordinator batch contract from Task 5 remains the distributed owner implementation and test boundary; do not label a local service call as a new remote transport.

Step 6: Verify vector routes and commit

rustfmt +nightly-2026-04-03 --edition 2021 --config skip_children=true src/apps/hnsw/mod.rs
cargo +nightly-2026-04-03 test --lib apps::hnsw::tests::cagra_vector_batch
cargo +nightly-2026-04-03 test --lib apps::hnsw::tests
git add src/apps/hnsw/mod.rs
git commit -m "feat(vector): route CAGRA searches through batching"

Task 8: Add Explicit Embedding Batch Search

Files:

Modify in Nebuchadnezzar: src/index/embedding/mod.rs
Modify in Morpheus: src/apps/embedding/mod.rs
Test in Morpheus: src/apps/embedding/tests.rs
Step 1: Add a failing Neb embedding default-adapter test

Create this recording core in the Neb module tests:

struct RecordingEmbeddingCore;

impl EmbeddingIndexerCore for RecordingEmbeddingCore {
    fn list_models(&self) -> BoxFuture<'_, Result<Vec<EmbeddingModelInfo>, IndexError>> {
        Box::pin(async { Ok(Vec::new()) })
    }

    fn insert(
        &self,
        _cell_id: &Id,
        _schema_id: u32,
        _field_id: u64,
        _model: &EmbeddingModel,
        _text: &str,
    ) -> BoxFuture<'_, Result<(), IndexError>> {
        Box::pin(async { Ok(()) })
    }

    fn remove(&self, _cell_id: &Id, _schema_id: u32, _field_id: u64)
        -> BoxFuture<'_, Result<(), IndexError>>
    {
        Box::pin(async { Ok(()) })
    }

    fn search(
        &self,
        _schema_id: u32,
        _field_id: u64,
        query: &str,
        _limit: usize,
    ) -> BoxFuture<'_, Result<Vec<EmbeddingHit>, IndexError>> {
        let score = query.len() as f32;
        Box::pin(async move { Ok(vec![EmbeddingHit { id: Id::new(0, score as u64), score }]) })
    }

    fn new_index(
        &self,
        _schema_id: u32,
        _field_id: u64,
        _model: &EmbeddingModel,
        _vector_config: VectorIndexConfig,
    ) -> BoxFuture<'_, Result<(), IndexError>> {
        Box::pin(async { Ok(()) })
    }

    fn delete_index(&self, _schema_id: u32, _field_id: u64)
        -> BoxFuture<'_, Result<(), IndexError>>
    {
        Box::pin(async { Ok(()) })
    }
}

Then assert:

let results = core
    .search_batch(
        1,
        2,
        vec!["third".to_string(), "one".to_string()],
        3,
    )
    .await
    .unwrap();
assert_eq!(results.len(), 2);
assert_eq!(results[0][0].score, 5.0);
assert_eq!(results[1][0].score, 3.0);

let error = core
    .search_batch(1, 2, Vec::new(), 3)
    .await
    .expect_err("empty embedding batch must fail");
assert!(error.to_string().contains("must not be empty"));

The test uses RecordingEmbeddingCore directly so it exercises the trait default before client forwarding is added.

Step 2: Add the Neb trait default and client method

Add:

fn search_batch(
    &self,
    schema_id: u32,
    field_id: u64,
    queries: Vec<String>,
    limit: usize,
) -> BoxFuture<'_, Result<Vec<Vec<EmbeddingHit>>, IndexError>> {
    Box::pin(async move {
        if queries.is_empty() {
            return Err(IndexError::Other(
                "embedding search batch must not be empty".to_string(),
            ));
        }
        let mut results = Vec::with_capacity(queries.len());
        for query in queries {
            results.push(self.search(schema_id, field_id, query.as_str(), limit).await?);
        }
        Ok(results)
    })
}

Add the owned forwarding method to EmbeddingIndexClient and verify:

rustfmt --edition 2021 --config skip_children=true src/index/embedding/mod.rs
cargo test embedding_batch --lib
git add src/index/embedding/mod.rs
git commit -m "feat(embedding): add ordered batch search API"

Step 3: Write a Morpheus provider/vector batching test

Extend the existing deterministic CUDA embedding fixture with query: gamma and a counted batch override:

struct DeterministicEmbeddingProvider {
    query_batch_calls: Arc<AtomicUsize>,
}

#[async_trait]
impl EmbeddingProvider for DeterministicEmbeddingProvider {
    async fn embed_queries_batch(
        &self,
        model: &str,
        texts: &[&str],
    ) -> Result<Vec<EmbeddingResult>, String> {
        self.query_batch_calls.fetch_add(1, Ordering::SeqCst);
        let mut results = Vec::with_capacity(texts.len());
        for text in texts {
            results.push(self.embed_query(model, text).await?);
        }
        Ok(results)
    }
}

In cagra_cuda_embedding_pipeline_e2e, retain the counter before registering the provider, call EmbeddingIndexerCore::search_batch with alpha then gamma, and assert:

assert_eq!(embedding_batches.len(), 2);
assert_eq!(embedding_batches[0][0].id, alpha_doc);
assert_eq!(embedding_batches[1][0].id, gamma_doc);
assert_eq!(query_batch_calls.load(Ordering::SeqCst), 1);

Add a source-contract unit test asserting the Morpheus override calls embed_queries_batch and then either search_batch or search_distributed_batch; Task 7's executor-count test proves the vector batch call is one CAGRA launch.

Step 4: Extract shared search resolution and hit mapping

Move the current model lookup into this async helper:

async fn resolve_search_index(
    &self,
    schema_id: u32,
    field_id: u64,
) -> Result<(String, u32, VectorIndexConfig), IndexError> {
    let configured_model = self
        .index_models
        .as_ref()
        .get(&(schema_id, field_id))
        .and_then(|models| models.read().first().cloned());
    if let Some(model) = configured_model {
        let (emb_schema_id, _, vector_config) = self
            .get_index_info(schema_id, field_id, &model)
            .await
            .map_err(|_| {
                IndexError::Other(format!(
                    "No embedding index found for schema {schema_id} field {field_id} (model: {model})"
                ))
            })?;
        return Ok((model, emb_schema_id, vector_config));
    }

    let default_full = Self::full_model_name(&self.default_provider, &self.default_model);
    let candidates = vec![
        default_full,
        "ollama:nomic-embed-text".to_string(),
        "gguf:multilingual-e5-base".to_string(),
        "e5:multilingual-e5-base".to_string(),
    ];
    for model in &candidates {
        if let Ok((emb_schema_id, _, vector_config)) =
            self.get_index_info(schema_id, field_id, model).await
        {
            return Ok((model.clone(), emb_schema_id, vector_config));
        }
    }
    Err(IndexError::Other(format!(
        "No embedding index found for schema {schema_id} field {field_id} (tried: {candidates:?})"
    )))
}

Move cell-ID mapping into:

async fn map_embedding_hits(
    &self,
    vector_config: VectorIndexConfig,
    hits: Vec<VectorHit>,
) -> Vec<EmbeddingHit> {
    let mut results = Vec::with_capacity(hits.len());
    for hit in hits {
        if let Ok(Ok(cell)) = self.neb_client.read_cell(hit.id).await {
            if let Some(document_id) = cell[DOC_CELL_ID].id() {
                results.push(EmbeddingHit {
                    id: *document_id,
                    score: embedding_hit_score(vector_config, hit.score),
                });
            }
        }
    }
    results
}

Refactor the single method to use these helpers without changing behavior.

Step 5: Override Morpheus embedding batch search

Resolve model/configuration once, split the provider/model name once, call embed_queries_batch once with Vec<&str>, require the provider result count to equal input count, derive ef_search once, then call vector search_batch or search_distributed_batch once. Require vector result count to match input count and map each hit list in input order.

fn search_batch(
    &self,
    schema_id: u32,
    field_id: u64,
    queries: Vec<String>,
    limit: usize,
) -> BoxFuture<'_, Result<Vec<Vec<EmbeddingHit>>, IndexError>> {
    async move {
        if queries.is_empty() {
            return Err(IndexError::Other(
                "embedding search batch must not be empty".to_string(),
            ));
        }
        let (model, emb_schema_id, vector_config) =
            self.resolve_search_index(schema_id, field_id).await?;
        let (provider_name, model_name) = model.split_once(':').ok_or_else(|| {
            IndexError::Other(format!("Invalid model format: {model}"))
        })?;
        let provider = self.get_provider(provider_name).map_err(IndexError::Other)?;
        let query_refs = queries.iter().map(String::as_str).collect::<Vec<_>>();
        let embeddings = provider
            .embed_queries_batch(model_name, query_refs.as_slice())
            .await
            .map_err(IndexError::Other)?;
        if embeddings.len() != queries.len() {
            return Err(IndexError::Other(format!(
                "embedding provider returned {} vectors for {} queries",
                embeddings.len(),
                queries.len()
            )));
        }
        let vectors = embeddings.into_iter().map(|result| result.vector).collect();
        let ef_search = search_ef_for_vector_config(limit, vector_config);
        let vector_hits = if self.vector_client.is_vector_search_coordinator_set() {
            self.vector_client
                .search_distributed_batch(
                    emb_schema_id,
                    vector_field_id(),
                    vectors,
                    limit,
                    ef_search,
                )
                .await?
        } else {
            self.vector_client
                .search_batch(emb_schema_id, vector_field_id(), vectors, limit, ef_search)
                .await?
        };
        if vector_hits.len() != queries.len() {
            return Err(IndexError::Other(format!(
                "vector batch returned {} results for {} embedding queries",
                vector_hits.len(),
                queries.len()
            )));
        }
        let mut results = Vec::with_capacity(vector_hits.len());
        for hits in vector_hits {
            results.push(self.map_embedding_hits(vector_config, hits).await);
        }
        Ok(results)
    }
    .boxed()
}

Step 6: Verify and commit embedding integration

rustfmt +nightly-2026-04-03 --edition 2021 --config skip_children=true src/apps/embedding/mod.rs src/apps/embedding/tests.rs
cargo +nightly-2026-04-03 test --lib embedding_batch
cargo +nightly-2026-04-03 test --lib apps::embedding::tests
git add src/apps/embedding/mod.rs src/apps/embedding/tests.rs
git diff --cached --check
git commit -m "feat(embedding): batch CAGRA query execution"

Task 9: Replace Batch Query Flattening With Reusable Pinned Staging

Files:

Modify: src/apps/cagra/gpu/cuda_oxide/search.rs
Modify: src/apps/cagra/gpu/tests.rs
Step 1: Write source-contract and behavioral tests

Add a source test asserting CagraCudaSearchWorkspace contains h_query: PinnedHostBuffer<f32>, batch launch calls copy_from_pinned_host_async, and flatten_search_queries is absent. Retain the existing multi-query CPU equivalence test as the behavioral gate.

#[test]
fn cagra_cuda_batch_uses_reusable_pinned_query_staging() {
    let source = include_str!("cuda_oxide/search.rs");
    assert!(source.contains("h_query: PinnedHostBuffer<f32>"));
    assert!(source.contains("copy_from_pinned_host_async(&stream, &workspace.h_query)"));
    assert!(!source.contains("fn flatten_search_queries("));
}

Step 2: Run the source test and observe the pageable path

cargo +nightly-2026-04-03 test --lib cagra_cuda_batch_uses_reusable_pinned_query_staging

Expected: the source-contract test fails because the workspace has no pinned query field.

Step 3: Allocate query staging with the workspace

Add h_query: PinnedHostBuffer<f32> next to d_query, allocate total_query_len in CagraCudaSearchWorkspace::new, map allocation errors with the segment ID, and include it in Self.

Step 4: Pack directly and launch from pinned memory

Change cagra_cuda_search_launch_batch to accept queries: &[Vec<f32>]. After validating shape, pack directly into the reusable buffer:

let mut offset = 0;
for query in queries {
    let end = offset + query_len;
    workspace.h_query.as_mut_slice()[offset..end].copy_from_slice(query);
    offset = end;
}
if let Err(error) = unsafe {
    workspace
        .d_query
        .copy_from_pinned_host_async(&stream, &workspace.h_query)
} {
    let _ = stream.synchronize();
    return Err(CagraError::Storage(format!(
        "failed to upload CAGRA cuda-oxide batch queries for segment {}: {error}",
        resident.key().segment_id
    )));
}

The pinned buffer remains owned by the workspace until the existing stream synchronization completes. Remove flatten_search_queries and its per-call Vec<f32> allocation.

Step 5: Verify correctness and memory safety on A4000

rustfmt +nightly-2026-04-03 --edition 2021 src/apps/cagra/gpu/cuda_oxide/search.rs src/apps/cagra/gpu/tests.rs
cargo +nightly-2026-04-03 test --features cagra-cuda --lib cagra_cuda_batch_uses_reusable_pinned_query_staging
cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_cuda_oxide_batch_search_smoke
compute-sanitizer --tool memcheck --error-exitcode=125 target/release/cagra_cuda_oxide_batch_search_smoke

Expected: the source contract and batch smoke pass, and memcheck reports zero errors.

Step 6: Measure the pinned-staging gate

Run the profile at dimensions 64, 384, and 768 with 64-query resident batches, five warmups, and 30 measured iterations. Record median per-query total, H2D, kernel, and D2H times. Reject the change if any dimension regresses by more than 3%; otherwise commit:

for dim in 64 384 768; do
  CAGRA_CUDA_PROFILE_ROWS=4096 CAGRA_CUDA_PROFILE_DIM="$dim" CAGRA_CUDA_PROFILE_DEGREE=32 CAGRA_CUDA_PROFILE_K=10 CAGRA_CUDA_PROFILE_SEARCH_WIDTH=32 CAGRA_CUDA_PROFILE_MAX_ITERATIONS=64 CAGRA_CUDA_PROFILE_QUERIES=64 CAGRA_CUDA_PROFILE_BATCH_SIZE=64 CAGRA_CUDA_PROFILE_WARMUP=5 CAGRA_CUDA_PROFILE_ITERS=30 cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_cuda_oxide_profile
done

git add src/apps/cagra/gpu/cuda_oxide/search.rs src/apps/cagra/gpu/tests.rs
git commit -m "perf(cagra): reuse pinned CUDA query staging"

Task 10: Add Reproducible Microbatch And Embedding GPU E2E Proof

Files:

Create: src/bin/cagra_microbatch_profile.rs
Modify: src/bin/cagra_cuda_oxide_embedding_e2e.rs
Step 1: Add parser tests for the microbatch profile

Define CLI and environment fields rows, dim, degree, queries, concurrency, warmup, iters, and device. Environment names are CAGRA_MICROBATCH_PROFILE_ROWS, DIM, DEGREE, QUERIES, CONCURRENCY, WARMUP, ITERS, and DEVICE; environment values take effect before CLI overrides. Test defaults of 4096, 64, 32, 640, 64, 5, 30, and device None, and reject concurrency zero or rows above CAGRA_CUDA_SEARCH_MAX_ROWS.

#[test]
fn profile_defaults_target_the_a4000_hard_workload() {
    let config = ProfileConfig::parse_from(Vec::<String>::new()).unwrap();
    assert_eq!(config.rows, 4096);
    assert_eq!(config.dim, 64);
    assert_eq!(config.degree, 32);
    assert_eq!(config.queries, 640);
    assert_eq!(config.concurrency, 64);
    assert_eq!(config.warmup, 5);
    assert_eq!(config.iters, 30);
    assert_eq!(config.device, None);
}

#[test]
fn profile_rejects_invalid_concurrency_and_rows() {
    assert!(ProfileConfig::parse_from(["--concurrency", "0"]).is_err());
    let rows = (CAGRA_CUDA_SEARCH_MAX_ROWS + 1).to_string();
    assert!(ProfileConfig::parse_from(vec!["--rows".to_string(), rows]).is_err());
}

Step 2: Implement direct, explicit, and transparent phases

Create one portable 4,096-row degree-32 base segment using CagraIndexService::compact_active_delta. Use RequireCuda for CUDA phases and Cpu for a repeated-single reference. Warm each execution thread independently: run one direct search on the main thread and one batcher search on its dedicated worker before timing, because cuda-oxide resident/module/workspace caches are thread-local. Report CSV columns:

phase,iteration,queries,batch_size,total_ms,ms_per_query,qps,p50_queue_us,p95_queue_us

For the transparent phase, submit concurrency compatible CagraSearchBatcher::search futures together and preserve each query/result association. Assert every phase returns the same ordered hits as repeated CPU singles before reporting performance.

Install CagraSearchBatcher::new_with_observer for the transparent phase. Append every queue_wait_us_by_query sample to a profile-local Mutex<Vec<u64>>, sort a copied vector after each measured iteration, and compute nearest-rank p50/p95. Assert:

assert!(microbatch_qps >= serialized_cuda_qps * 1.5);
assert!(queue_p95 <= batching_config.max_delay_us + slowest_batch_execution_us);

Step 3: Extend embedding E2E to use the explicit batch API

Add deterministic vectors for query alpha and query gamma, then call:

let batch_hits = EmbeddingIndexerCore::search_batch(
    &indexer,
    DOC_SCHEMA_ID,
    doc_field_id,
    vec!["query alpha".to_string(), "query gamma".to_string()],
    2,
)
.await
.map_err(|error| other_error(format!("embedding batch search failed: {error:?}")))?;

Assert two result lists, alpha first in the first list, gamma first in the second list, and the provider's batch method was called once. Keep RequireCuda and the runtime probe.

Step 4: Run A4000 E2E and throughput gates

cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_cuda_oxide_embedding_e2e
CAGRA_MICROBATCH_PROFILE_ROWS=4096 CAGRA_MICROBATCH_PROFILE_DIM=64 CAGRA_MICROBATCH_PROFILE_DEGREE=32 CAGRA_MICROBATCH_PROFILE_QUERIES=640 CAGRA_MICROBATCH_PROFILE_CONCURRENCY=64 CAGRA_MICROBATCH_PROFILE_WARMUP=5 CAGRA_MICROBATCH_PROFILE_ITERS=30 cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_microbatch_profile

Expected: embedding vector and text batch results are correct, explicit CUDA batch is no slower than 0.597 ms/query, and transparent compatible singles achieve at least 1.5x serialized direct-single QPS.

Step 5: Commit reproducible proof binaries

git add src/bin/cagra_microbatch_profile.rs src/bin/cagra_cuda_oxide_embedding_e2e.rs
git commit -m "test(cagra): prove batched GPU query pipelines"

Task 11: Run The Shared-Memory Query Cache Experiment

Files:

Modify conditionally: src/apps/cagra/gpu/cuda_common.rs
Modify conditionally: src/apps/cagra/gpu/cuda_oxide/search.rs
Modify conditionally: src/apps/cagra/gpu/mod.rs
Modify conditionally: src/apps/cagra/gpu/tests.rs
Modify conditionally: src/bin/cagra_cuda_oxide_batch_search_smoke.rs
Modify conditionally: src/bin/cagra_cuda_oxide_profile.rs
Step 1: Capture an Nsight Compute baseline before changing the kernel

CAGRA_CUDA_PROFILE_ROWS=4096 CAGRA_CUDA_PROFILE_DIM=64 CAGRA_CUDA_PROFILE_DEGREE=32 CAGRA_CUDA_PROFILE_K=10 CAGRA_CUDA_PROFILE_SEARCH_WIDTH=32 CAGRA_CUDA_PROFILE_MAX_ITERATIONS=64 CAGRA_CUDA_PROFILE_QUERIES=64 CAGRA_CUDA_PROFILE_BATCH_SIZE=64 CAGRA_CUDA_PROFILE_WARMUP=5 CAGRA_CUDA_PROFILE_ITERS=20 cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_cuda_oxide_profile
ncu --set full --kernel-name regex:cagra_search_l2_oxide_batch --launch-count 10 --target-processes all --export /tmp/cagra-batch-global-query.ncu-rep target/release/cagra_cuda_oxide_profile --rows 4096 --dim 64 --degree 32 --k 10 --search-width 32 --max-iterations 64 --queries 64 --batch-size 64 --warmup 5 --iters 20
ncu --import /tmp/cagra-batch-global-query.ncu-rep --page raw --csv > /tmp/cagra-batch-global-query.csv

Record kernel duration, achieved occupancy, registers per thread, dynamic shared memory per block, L1/TEX hit rate, L2 hit rate, and DRAM bytes.

Step 2: Write a failing cached/global parity test

Add a host-only launch override used by tests and the profile:

#[doc(hidden)]
pub fn search_base_segment_graph_cuda_batch_with_query_cache_for_tests(
    schema_id: u32,
    field_id: u64,
    descriptor: &CagraSegmentDescriptor,
    segment: &CagraBaseSegment,
    queries: &[Vec<f32>],
    config: CagraSearchConfig,
    preferred_device_ordinal: Option<u32>,
    cache_query: bool,
) -> Result<Vec<CagraBackendSearchResult>, CagraError>

Add the matching #[doc(hidden)] pub wrapper in src/apps/cagra/gpu/mod.rs; the underlying cuda-oxide function remains pub(crate). The wrapper delegates to the same production implementation while overriding only query-cache admission. Then run identical batches with caching disabled and enabled and assert exact cell-ID order and distance tolerance 1e-5 for dimensions 64, 384, and 768.

Add CAGRA_CUDA_PROFILE_QUERY_CACHE=0|1 parsing to cagra_cuda_oxide_profile.rs. When present, call the public profiled facade above; when absent, call the production batch facade. Include query_cache=global|shared|production in the summary line.

Add a host source contract that regular cargo test can execute without launching CUDA:

#[test]
fn cagra_cuda_batch_query_cache_source_contract() {
    let source = include_str!("cuda_oxide/search.rs");
    assert!(source.contains("query_cache_slots"));
    assert!(source.contains("DynamicSharedArray::<u32>::get()"));
    assert!(source.contains("search_base_segment_graph_cuda_batch_with_query_cache_for_tests"));
}

#[cfg(feature = "cagra-cuda")]
#[test]
#[ignore = "executed through cagra_cuda_oxide_batch_search_smoke"]
fn cagra_cuda_batch_query_cache_matches_global() {
    for dimension in [64_usize, 384, 768] {
        let row_count = 8;
        let segment = CagraBaseSegment {
            artifact_version: CAGRA_ARTIFACT_VERSION,
            segment_id: 700 + dimension as u64,
            generation: 1,
            dimension: dimension as u16,
            vectors: (0..row_count)
                .flat_map(|row| (0..dimension).map(move |column| (row + column) as f32 / 17.0))
                .collect(),
            cell_ids: (0..row_count).map(|row| Id::new(0, row as u64 + 1)).collect(),
            adjacency: complete_graph_adjacency(row_count),
        };
        let descriptor = sample_base_descriptor(&segment);
        let queries = vec![segment.vectors[..dimension].to_vec(), vec![0.25; dimension]];
        let config = CagraSearchConfig { k: 4, search_width: 4, max_iterations: 16 };
        let global = search_base_segment_graph_cuda_batch_with_query_cache_for_tests(
            1, 2, &descriptor, &segment, &queries, config, None, false,
        )
        .unwrap();
        let cached = search_base_segment_graph_cuda_batch_with_query_cache_for_tests(
            1, 2, &descriptor, &segment, &queries, config, None, true,
        )
        .unwrap();
        for (global_query, cached_query) in global.iter().zip(&cached) {
            assert_eq!(
                global_query.hits.iter().map(|hit| hit.cell_id).collect::<Vec<_>>(),
                cached_query.hits.iter().map(|hit| hit.cell_id).collect::<Vec<_>>()
            );
            for (left, right) in global_query.hits.iter().zip(&cached_query.hits) {
                assert!((left.distance - right.distance).abs() <= 1e-5);
            }
        }
    }
}

Implement the test-only facade by threading a cache_query: bool host parameter into cagra_cuda_search_launch_batch; production derives the value from the accepted default after Step 6.

Add this device loop to cagra_cuda_oxide_batch_search_smoke.rs so cuda-oxide compiles and executes both variants on the A4000:

fn run_query_cache_parity() {
    for dimension in [64_usize, 384, 768] {
        let row_count = 8;
        let vectors = (0..row_count)
            .flat_map(|row| (0..dimension).map(move |column| (row + column) as f32 / 17.0))
            .collect::<Vec<_>>();
        let cell_ids = (0..row_count)
            .map(|row| Id::new(0, row as u64 + 1))
            .collect::<Vec<_>>();
        let segment = build_base_segment_exact_l2_cuda(
            800 + dimension as u64,
            1,
            dimension as u16,
            &vectors,
            &cell_ids,
            4,
            None,
            None,
        )
        .unwrap();
        let descriptor = base_segment_descriptor(&segment).unwrap();
        let queries = vec![vectors[..dimension].to_vec(), vec![0.25; dimension]];
        let config = CagraSearchConfig { k: 4, search_width: 4, max_iterations: 16 };
        let global = search_base_segment_graph_cuda_batch_with_query_cache_for_tests(
            1, 2, &descriptor, &segment, &queries, config, None, false,
        )
        .unwrap();
        let cached = search_base_segment_graph_cuda_batch_with_query_cache_for_tests(
            1, 2, &descriptor, &segment, &queries, config, None, true,
        )
        .unwrap();
        assert_batch_results_close(&global, &cached, 1e-5);
    }
}

Use this comparison helper:

fn assert_batch_results_close(
    left: &[CagraBackendSearchResult],
    right: &[CagraBackendSearchResult],
    tolerance: f32,
) {
    assert_eq!(left.len(), right.len());
    for (left_query, right_query) in left.iter().zip(right) {
        assert_eq!(left_query.hits.len(), right_query.hits.len());
        for (left_hit, right_hit) in left_query.hits.iter().zip(&right_query.hits) {
            assert_eq!(left_hit.cell_id, right_hit.cell_id);
            assert!((left_hit.distance - right_hit.distance).abs() <= tolerance);
        }
    }
}

Keep the host unit test as a source/type contract and the smoke as the device correctness contract.

Step 3: Reserve query bytes without violating hash-table admission

Add a batch-specific resource helper. When caching is enabled and dimension * size_of::<f32>() leaves at least one valid hash-table slot, call cagra_cuda_search_launch_resources with the remaining budget, then add query bytes to shared_mem_bytes. Otherwise set query_cache_slots to zero and retain the global path.

struct CagraCudaBatchLaunchResources {
    search: CagraCudaSearchLaunchResources,
    query_cache_slots: usize,
    shared_mem_bytes: u32,
}

fn cagra_cuda_batch_launch_resources(
    row_count: usize,
    graph_degree: usize,
    dimension: usize,
    config: CagraSearchConfig,
    dynamic_shared_budget_bytes: usize,
    cache_query: bool,
) -> Result<CagraCudaBatchLaunchResources, CagraError> {
    let global = || {
        let search = cagra_cuda_search_launch_resources(
            row_count,
            graph_degree,
            config,
            dynamic_shared_budget_bytes,
        )?;
        Ok(CagraCudaBatchLaunchResources {
            shared_mem_bytes: search.shared_mem_bytes,
            search,
            query_cache_slots: 0,
        })
    };
    if !cache_query {
        return global();
    }
    let query_bytes = dimension.checked_mul(size_of::<f32>()).ok_or_else(|| {
        CagraError::InvalidConfig("query cache byte count overflows".to_string())
    })?;
    let Some(hash_budget) = dynamic_shared_budget_bytes.checked_sub(query_bytes) else {
        return global();
    };
    if hash_budget < size_of::<u32>() {
        return global();
    }
    let search = cagra_cuda_search_launch_resources(
        row_count,
        graph_degree,
        config,
        hash_budget,
    )?;
    let shared_mem_bytes = u32::try_from(query_bytes + search.shared_mem_bytes as usize)
        .map_err(|_| CagraError::InvalidConfig("batch shared memory exceeds u32".to_string()))?;
    Ok(CagraCudaBatchLaunchResources {
        search,
        query_cache_slots: dimension,
        shared_mem_bytes,
    })
}

Step 4: Partition dynamic shared memory in the batch kernel

Add query_cache_slots: u32 to the batch kernel. Treat the first slots as f32 query values and the remaining slots as the existing u32 hash table. Cooperatively copy one query per CTA and synchronize before traversal:

let shared = DynamicSharedArray::<u32>::get();
let query_cache = shared.cast::<f32>();
let hash_table = unsafe { shared.add(query_cache_slots as usize) };
let query_base = (query_index as usize) * (dimension as usize);
let mut dim = thread::threadIdx_x();
while dim < query_cache_slots {
    unsafe {
        *query_cache.add(dim as usize) = queries[query_base + dim as usize];
    }
    dim += thread::blockDim_x();
}
thread::sync_threads();

Pass the cache pointer and an enabled flag through traversal and both squared-L2 helpers. Read from the cache when enabled and from queries[query_base + dim_index] otherwise. Single-query kernels keep the global path.

Step 5: Run correctness and all three sanitizers

cargo +nightly-2026-04-03 test --features cagra-cuda --lib cagra_cuda_batch_query_cache_source_contract
cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_cuda_oxide_batch_search_smoke
compute-sanitizer --tool memcheck --error-exitcode=125 target/release/cagra_cuda_oxide_batch_search_smoke
compute-sanitizer --tool racecheck --error-exitcode=125 target/release/cagra_cuda_oxide_batch_search_smoke
compute-sanitizer --tool synccheck --error-exitcode=125 target/release/cagra_cuda_oxide_batch_search_smoke

Expected: exact ordered IDs, distance tolerance satisfied, and zero sanitizer errors.

Step 6: Benchmark both variants and apply the retention rule

For dimensions 64, 384, and 768, run five warmups and 30 iterations with caching forced off and on. Re-run Nsight Compute for the enabled variant. Retain the implementation only if the hard dimension-64 resident median improves by at least 5%, no tested dimension regresses by more than 3%, and occupancy loss does not erase QPS. If the gate fails, remove the experimental code with a reviewable patch and keep the measured CSV/reports outside git; do not rewrite history.

for cache in 0 1; do
  for dim in 64 384 768; do
    CAGRA_CUDA_PROFILE_QUERY_CACHE="$cache" CAGRA_CUDA_PROFILE_ROWS=4096 CAGRA_CUDA_PROFILE_DIM="$dim" CAGRA_CUDA_PROFILE_DEGREE=32 CAGRA_CUDA_PROFILE_K=10 CAGRA_CUDA_PROFILE_SEARCH_WIDTH=32 CAGRA_CUDA_PROFILE_MAX_ITERATIONS=64 CAGRA_CUDA_PROFILE_QUERIES=64 CAGRA_CUDA_PROFILE_BATCH_SIZE=64 CAGRA_CUDA_PROFILE_WARMUP=5 CAGRA_CUDA_PROFILE_ITERS=30 cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_cuda_oxide_profile
  done
done

Step 7: Commit only a retained optimization

git add src/apps/cagra/gpu/cuda_common.rs src/apps/cagra/gpu/cuda_oxide/search.rs src/apps/cagra/gpu/mod.rs src/apps/cagra/gpu/tests.rs src/bin/cagra_cuda_oxide_batch_search_smoke.rs src/bin/cagra_cuda_oxide_profile.rs
git commit -m "perf(cagra): cache batched queries in shared memory"

If the retention gate fails, skip this commit and record the rejection and measurements in the final implementation report.

Task 12: Full Verification, Review, And Final Commit Audit

Files:

Verify all files listed above
Step 1: Format and compile both repositories

Nebuchadnezzar:

cargo fmt --check
cargo test index::vector::tests --lib
cargo test index::embedding::tests --lib

Morpheus:

cargo +nightly-2026-04-03 fmt --check
cargo +nightly-2026-04-03 check --lib
cargo +nightly-2026-04-03 check --lib --features cagra-cuda

Expected: formatting and compilation pass.

Step 2: Run CPU, distributed, microbatch, and embedding suites

cargo +nightly-2026-04-03 test --lib cagra_batch
cargo +nightly-2026-04-03 test --lib cagra_service_batch
cargo +nightly-2026-04-03 test --lib cagra_distributed_batch
cargo +nightly-2026-04-03 test --lib apps::cagra::batch
cargo +nightly-2026-04-03 test --lib embedding_batch
cargo +nightly-2026-04-03 test --lib apps::hnsw::tests

Expected: every focused suite passes.

Step 3: Run final A4000 CUDA correctness and sanitizer gates

cargo +nightly-2026-04-03 test --features cagra-cuda --lib apps::cagra::gpu::tests::cuda
cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_cuda_oxide_correctness
cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_cuda_oxide_batch_search_smoke
cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_cuda_oxide_embedding_e2e
compute-sanitizer --tool memcheck --error-exitcode=125 target/release/cagra_cuda_oxide_batch_search_smoke
compute-sanitizer --tool racecheck --error-exitcode=125 target/release/cagra_cuda_oxide_batch_search_smoke
compute-sanitizer --tool synccheck --error-exitcode=125 target/release/cagra_cuda_oxide_batch_search_smoke

Expected: CUDA tests and embedding E2E pass; each sanitizer reports zero errors.

Step 4: Run final performance acceptance profiles

CAGRA_CUDA_PROFILE_ROWS=4096 CAGRA_CUDA_PROFILE_DIM=64 CAGRA_CUDA_PROFILE_DEGREE=32 CAGRA_CUDA_PROFILE_K=10 CAGRA_CUDA_PROFILE_SEARCH_WIDTH=32 CAGRA_CUDA_PROFILE_MAX_ITERATIONS=64 CAGRA_CUDA_PROFILE_QUERIES=64 CAGRA_CUDA_PROFILE_BATCH_SIZE=64 CAGRA_CUDA_PROFILE_WARMUP=5 CAGRA_CUDA_PROFILE_ITERS=30 cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_cuda_oxide_profile
CAGRA_MICROBATCH_PROFILE_ROWS=4096 CAGRA_MICROBATCH_PROFILE_DIM=64 CAGRA_MICROBATCH_PROFILE_DEGREE=32 CAGRA_MICROBATCH_PROFILE_QUERIES=640 CAGRA_MICROBATCH_PROFILE_CONCURRENCY=64 CAGRA_MICROBATCH_PROFILE_WARMUP=5 CAGRA_MICROBATCH_PROFILE_ITERS=30 cargo +nightly-2026-04-03 oxide run --arch sm_86 --features cagra-cuda --bin cagra_microbatch_profile

Expected: explicit batch meets the 0.597 ms/query ceiling and transparent microbatching meets the 1.5x QPS gate. Report cold and resident runs separately.

Step 5: Run two-stage review

Use superpowers:requesting-code-review first for spec compliance, then for code quality. Resolve every correctness, ownership, ordering, cancellation, and test-isolation finding before continuing. Re-run the smallest affected suite after each fix and then repeat Steps 1-4.

Step 6: Audit commit and worktree scope

git log --oneline --decorate -15
git status --short
git diff --check
git diff develop...HEAD --stat

Expected: batch commits are narrowly scoped, unrelated pre-existing dirty files remain untouched, no persistent CAGRA format changed, and no CUDA data is duplicated between cuda-oxide and cudarc.

Final Evidence To Report

Nebuchadnezzar and Morpheus commit IDs.
Focused CPU/distributed/microbatch/embedding test counts.
A4000 model, compute capability sm_86, driver, and CUDA toolkit versions.
Explicit batch and transparent microbatch QPS, median/p95 latency, and queue p95.
Cold versus resident H2D/kernel/D2H timing.
Compute Sanitizer memcheck/racecheck/synccheck summaries.
Whether shared-memory query caching was retained or rejected, with the measured percentage and occupancy reason.
Remaining declared limit: CUDA CAGRA V1 still admits at most 4,096 rows; this plan does not lift it.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

CAGRA Batched Search And Microbatching Implementation Plan

Execution Boundaries

File Structure

Nebuchadnezzar

Morpheus

Acceptance Gates

Task 0: Preserve And Seal The Existing CUDA Baseline

Task 1: Add Nebuchadnezzar Vector Batch Contracts

Task 2: Add And Validate CAGRA Batching Configuration

Task 3: Add Portable Local And RPC Batch Contracts

Task 4: Implement Manifest-Pinned Service Batch Search

Task 5: Add Strict Distributed Batch Coordination

Task 6: Implement The Dedicated Transparent Microbatch Worker

Task 7: Route Vector Singles And Explicit Batches Through CAGRA

Task 8: Add Explicit Embedding Batch Search

Task 9: Replace Batch Query Flattening With Reusable Pinned Staging

Task 10: Add Reproducible Microbatch And Embedding GPU E2E Proof

Task 11: Run The Shared-Memory Query Cache Experiment

Task 12: Full Verification, Review, And Final Commit Audit

Final Evidence To Report

Uh oh!

FilesExpand file tree

CAGRA_BATCHED_SEARCH_IMPLEMENTATION_PLAN.md

Latest commit

History

CAGRA_BATCHED_SEARCH_IMPLEMENTATION_PLAN.md

File metadata and controls

CAGRA Batched Search And Microbatching Implementation Plan

Execution Boundaries

File Structure

Nebuchadnezzar

Morpheus

Acceptance Gates

Task 0: Preserve And Seal The Existing CUDA Baseline

Task 1: Add Nebuchadnezzar Vector Batch Contracts

Task 2: Add And Validate CAGRA Batching Configuration

Task 3: Add Portable Local And RPC Batch Contracts

Task 4: Implement Manifest-Pinned Service Batch Search

Task 5: Add Strict Distributed Batch Coordination

Task 6: Implement The Dedicated Transparent Microbatch Worker

Task 7: Route Vector Singles And Explicit Batches Through CAGRA

Task 8: Add Explicit Embedding Batch Search

Task 9: Replace Batch Query Flattening With Reusable Pinned Staging

Task 10: Add Reproducible Microbatch And Embedding GPU E2E Proof

Task 11: Run The Shared-Memory Query Cache Experiment

Task 12: Full Verification, Review, And Final Commit Audit

Final Evidence To Report