feat(block_manager): add DirectAccess for synchronous pool operations

Introduces a DirectAccess struct that provides synchronous access to the
block pool state, bypassing the progress engine for performance-critical paths.

This is a simplified initial implementation that provides basic direct access
without complex retry logic. The implementation includes:
- Direct block allocation from the pool
- Direct block addition to inactive pool
- Block return functionality
- Pool status and reset operations

The DirectAccess pattern allows for more efficient pool operations when
async coordination is not required.

Signed-off-by: Ryan Olson <rolson@nvidia.com>

Author: ryanolson

lib/llm/src/block_manager/pool/managed.rs

@@ -47,6 +47,7 @@ use super::*;
 
 pub mod active;
 pub mod controller;
+pub mod direct;
 pub mod inactive;
 pub mod priority_key;
 pub mod state;

lib/llm/src/block_manager/pool/managed/direct.rs

@@ -0,0 +1,79 @@
+use super::*;
+use crate::block_manager::pool::{BlockPoolError, BlockPoolResult, ResetBlocksResponse};
+use std::sync::{Arc, Mutex};
+
+/// Direct access to the block pool state, bypassing the progress engine.
+/// This provides synchronous access for performance-critical paths.
+///
+/// Note: This is a simplified initial implementation that provides basic
+/// direct access without complex retry logic.
+pub struct DirectAccess<S: Storage, L: LocalityProvider, M: BlockMetadata> {
+    state: Arc<Mutex<State<S, L, M>>>,
+}
+
+impl<S: Storage, L: LocalityProvider, M: BlockMetadata> Clone for DirectAccess<S, L, M> {
+    fn clone(&self) -> Self {
+        Self {
+            state: self.state.clone(),
+        }
+    }
+}
+
+impl<S: Storage, L: LocalityProvider, M: BlockMetadata> DirectAccess<S, L, M> {
+    pub fn new(state: Arc<Mutex<State<S, L, M>>>) -> Self {
+        Self { state }
+    }
+
+    /// Get a reference to the state - used for testing
+    #[allow(dead_code)]
+    pub(crate) fn state(&self) -> Arc<Mutex<State<S, L, M>>> {
+        self.state.clone()
+    }
+
+    /// Allocate a set of blocks from the pool.
+    pub fn allocate_blocks(&self, count: usize) -> BlockPoolResult<Vec<MutableBlock<S, L, M>>> {
+        let mut state = self.state.lock().unwrap();
+        state.allocate_blocks(count)
+    }
+
+    /// Add blocks to the inactive pool.
+    pub fn add_blocks(&self, blocks: Vec<Block<S, L, M>>) {
+        let mut state = self.state.lock().unwrap();
+        state.inactive.add_blocks(blocks);
+    }
+
+    /// Try to return a block to the pool.
+    pub fn try_return_block(&self, block: Vec<Block<S, L, M>>) -> BlockPoolResult<()> {
+        if block.is_empty() {
+            return Ok(());
+        }
+
+        let mut state = self.state.lock().unwrap();
+        for b in block {
+            state.return_block(b);
+        }
+
+        Ok(())
+    }
+
+    /// Get the current status of the block pool.
+    pub fn status(&self) -> Result<BlockPoolStatus, BlockPoolError> {
+        let state = self.state.lock().unwrap();
+        Ok(state.status())
+    }
+
+    /// Reset the pool, returning all blocks to the inactive state.
+    pub fn reset(&self) -> Result<(), BlockPoolError> {
+        let mut state = self.state.lock().unwrap();
+        state.inactive.reset()
+    }
+
+    /// Reset specific blocks by sequence hash.
+    pub fn reset_blocks(
+        &self,
+        sequence_hashes: &[SequenceHash],
+    ) -> Result<ResetBlocksResponse, BlockPoolError> {
+        let mut state = self.state.lock().unwrap();
+        Ok(state.try_reset_blocks(sequence_hashes))
+    }
+}

lib/llm/src/block_manager/pool/managed/state.rs

@@ -371,11 +371,11 @@ impl<S: Storage, L: LocalityProvider + 'static, M: BlockMetadata> State<S, L, M>
         self.inactive.return_block(block);
     }
 
-    fn publisher(&self) -> Publisher {
+    pub fn publisher(&self) -> Publisher {
         Publisher::new(self.event_manager.clone())
     }
 
-    fn status(&self) -> BlockPoolStatus {
+    pub fn status(&self) -> BlockPoolStatus {
         let active = self.active.status();
         let (inactive, empty) = self.inactive.status();
         BlockPoolStatus {
@@ -385,7 +385,7 @@ impl<S: Storage, L: LocalityProvider + 'static, M: BlockMetadata> State<S, L, M>
         }
     }
 
-    fn try_reset_blocks(&mut self, sequence_hashes: &[SequenceHash]) -> ResetBlocksResponse {
+    pub fn try_reset_blocks(&mut self, sequence_hashes: &[SequenceHash]) -> ResetBlocksResponse {
         let mut reset_blocks = Vec::new();
         let mut not_found = Vec::new();
         let mut not_reset = Vec::new();