providers/transports

• Schema normalization (Anthropic message blocks → unified content shape, OpenAI tool-call ids → canonical record).
• Reasoning-detail replay (OpenAI Responses reasoning items, Anthropic thinking blocks).
• Cache-marker placement (Anthropic cache_control, OpenAI prompt_cache_key).
• Token-field interpretation (cached-read vs cache-write deltas, reasoning tokens, audio tokens).
• Auth flow (API key, OAuth PKCE, device code).
• Model policy (variant aliases, structured-output capability detection, thinking exposure).

Modes (standard, rlm) and canonical tool definitions never see provider-specific JSON. They operate on LlmRequest, LlmResponse, and ToolDefinition only.

Provider specs remain host config data. Runtime sessions persist only provider ids and bind them to the single live ProviderHandle supplied by the host when the session opens.

Breakpoints are placed at:

1. The first system/developer text message.
2. The last tool definition in the request.
3. Any explicit LlmContentBlock::Text.cache_breakpoint the runtime asks for.

When no explicit breakpoint is set, the provider falls back to the last user/assistant text content so prompt caching still works for sessions without explicit cache instrumentation.

The result: long RLM sessions get the same cache-hit rate as native multi-turn chats, even though Lashlang-driven reasoning regenerates a fresh prompt on every iteration.

use lash::provider::{ProviderReliability, RequestTimeout};

let reliability = ProviderReliability::default()
    .request_timeout(Some(RequestTimeout::Millis(120_000)))
    .stream_chunk_timeout_ms(Some(30_000))
    .max_attempts(3)
    .max_concurrency(Some(8))
    .requests_per_window(Some(120), Some(60_000));

/// Host-owned admission: bounded in-flight windows per traffic class,
/// wrapped around the provider the host installs. Breakers, AIMD windows,
/// and backpressure metrics slot into the same `complete()` seam.
#[derive(Debug)]
struct AdmissionGate {
    inner: Box<dyn Provider>,
    interactive_slots: Arc<tokio::sync::Semaphore>,
    batch_slots: Arc<tokio::sync::Semaphore>,
}

impl Clone for AdmissionGate {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone_boxed(),
            // Shared handles: every clone admits through the same windows.
            interactive_slots: Arc::clone(&self.interactive_slots),
            batch_slots: Arc::clone(&self.batch_slots),
        }
    }
}

#[async_trait::async_trait]
impl Provider for AdmissionGate {
    async fn complete(&mut self, request: LlmRequest) -> Result<LlmResponse, LlmTransportError> {
        // Class traffic by the session identity the host already owns:
        // deliberate ids for its own pipelines, and a default lane for ids
        // it did not mint (lash-spawned child sessions).
        let lane = if request.scope.session_id.starts_with("batch:") {
            &self.batch_slots
        } else {
            &self.interactive_slots
        };
        // The permit drops on every exit path — success, failure, or a
        // cancelled turn — so an aborted call never leaks a slot.
        let _slot = lane.acquire().await.expect("admission gate closed");
        self.inner.complete(request).await
    }

    // Forward `close()` explicitly: the default impl is a no-op and would
    // silently skip the inner provider's transport shutdown.
    async fn close(&self) -> Result<(), LlmTransportError> {
        self.inner.close().await
    }

    fn kind(&self) -> &'static str {
        self.inner.kind()
    }
    fn options(&self) -> ProviderOptions {
        self.inner.options()
    }
    fn set_options(&mut self, options: ProviderOptions) {
        self.inner.set_options(options);
    }
    fn serialize_config(&self) -> serde_json::Value {
        self.inner.serialize_config()
    }
    fn requires_streaming(&self) -> bool {
        self.inner.requires_streaming()
    }
    fn clone_boxed(&self) -> Box<dyn Provider> {
        Box::new(self.clone())
    }
}

Install the wrapper with ProviderComponents::map_provider wherever the handle is constructed:

let interactive_slots = Arc::new(tokio::sync::Semaphore::new(8));
let batch_slots = Arc::new(tokio::sync::Semaphore::new(2));
let handle = ProviderHandle::new(components.map_provider(|inner| {
    Box::new(AdmissionGate {
        inner,
        interactive_slots,
        batch_slots,
    })
}));

The retry ladder wraps the decorator, so core stays honest about throttling on its own (see the throttles row above): an admission gate cannot compensate for a ladder that would turn a 429 storm into attempt exhaustion. The admission use-case pairs with the metrics guidance on operations.

pub struct LlmUsage {
    pub input_tokens: i64,              // Uncached ordinary input
    pub cache_read_input_tokens: i64,   // Prompt input read from cache
    pub cache_write_input_tokens: i64,  // Prompt input written to cache
    pub output_tokens: i64,             // Total generated output
    pub reasoning_output_tokens: i64,   // Subset of output tokens
    // …provider-specific extras flow through the extended trace
}

See lash-provider-openai as the most general template: it handles both the direct Responses path and the generic Chat Completions path in one crate.