soma_provider_adapters/

python.rs

//! The generic Python (/ LangChain / LlamaIndex) provider kind: introspects
//! and executes a drop-in `.py` provider through a bounded Python sidecar
//! running `python_bridge::python_bridge_program`. Ported from
//! `soma-service::providers::python`.
//!
//! `load_python_catalog` here applies only *generic* manifest validation
//! (`soma_provider_core::validate_provider_manifest_value`) — Soma's own CLI
//! reserved-command / env-prefix policy layer is applied downstream by the
//! host's provider registry when it builds every provider's catalog
//! (regardless of kind), so dropping the redundant Soma-specific pre-check
//! here does not weaken overall enforcement — see the PR10 deviation notes.

use std::{
    io::{Read, Write},
    path::{Path, PathBuf},
    process::{Command as StdCommand, Stdio as StdStdio},
    sync::Arc,
    thread,
    time::{Duration, Instant},
};

use async_trait::async_trait;
use serde_json::Value;
use soma_provider_core::{
    Provider, ProviderCall, ProviderCatalog, ProviderError, ProviderOutput, ProviderTool,
    validate_provider_manifest_value,
};
use tokio::time::Instant as TokioInstant;

use crate::{
    error::{SidecarError, redact_public},
    python_bridge::python_bridge_program,
    python_protocol::{
        PYTHON_PROTOCOL_HEADROOM_BYTES, PythonProtocolError, PythonWorkerRequest,
        PythonWorkerResponse, decode_python_response, encode_python_request,
        validate_python_response,
    },
    sidecar::{
        collect_provider_env, output_exceeded_message, resolve_sidecar_command,
        run_bounded_sidecar, sidecar_base_env, spawn_retrying_busy_image_blocking,
    },
};
use supervisor::{PythonSupervisorConfig, PythonWorkerIdentity, PythonWorkerSupervisor};

pub mod cache;
mod catalog;
mod containment;
pub mod environment;
pub mod host;
mod interpreter;
pub mod lifecycle;
pub mod materializer;
pub mod supervisor;
pub use interpreter::PythonInterpreter;
pub(crate) use interpreter::default_python_command;
use interpreter::select_python_command;

const DEFAULT_TIMEOUT_MS: u64 = 10_000;
const DEFAULT_MAX_INPUT_BYTES: usize = 64 * 1024;
const DEFAULT_MAX_OUTPUT_BYTES: usize = 256 * 1024;

pub use catalog::describe_persistent_catalog;

fn protocol_output_limit(payload_limit: usize) -> usize {
    payload_limit.saturating_add(PYTHON_PROTOCOL_HEADROOM_BYTES)
}

#[derive(Clone)]
pub struct PythonProvider {
    path: PathBuf,
    catalog: ProviderCatalog,
    env_prefix: String,
    interpreter: PythonInterpreter,
    supervisor: Option<Arc<PythonWorkerSupervisor>>,
}

impl PythonProvider {
    /// Starts, negotiates, describes, and health-checks a persistent worker.
    /// One-shot providers have no persistent preflight and return immediately.
    pub async fn preflight(&self) -> Result<(), ProviderError> {
        let Some(supervisor) = &self.supervisor else {
            return Ok(());
        };
        supervisor.preflight().await.map(|_| ()).map_err(|error| {
            ProviderError::new(
                error.code(),
                &self.catalog.provider.name,
                None,
                error.to_string(),
                "Install the matching soma-provider wheel and inspect the provider source.",
            )
            .with_provider_kind(self.catalog.provider.kind.as_str())
            .with_source(self.path.display().to_string())
            .with_phase("persistent-preflight")
        })
    }

    pub fn new(path: PathBuf, catalog: ProviderCatalog, env_prefix: impl Into<String>) -> Self {
        Self::new_with_interpreter(path, catalog, env_prefix, PythonInterpreter::Ambient)
    }

    pub fn new_with_interpreter(
        path: PathBuf,
        catalog: ProviderCatalog,
        env_prefix: impl Into<String>,
        interpreter: PythonInterpreter,
    ) -> Self {
        Self {
            path,
            catalog,
            env_prefix: env_prefix.into(),
            interpreter,
            supervisor: None,
        }
    }

    /// Builds a provider backed by one lazily-started persistent worker.
    pub fn new_persistent(
        path: PathBuf,
        catalog: ProviderCatalog,
        env_prefix: impl Into<String>,
        interpreter: PythonInterpreter,
        config: PythonSupervisorConfig,
    ) -> Result<Self, ProviderError> {
        Self::new_persistent_inner(path, catalog, env_prefix, interpreter, config, None)
    }

    fn new_persistent_inner(
        path: PathBuf,
        catalog: ProviderCatalog,
        env_prefix: impl Into<String>,
        interpreter: PythonInterpreter,
        config: PythonSupervisorConfig,
        immutable_generation_digest: Option<String>,
    ) -> Result<Self, ProviderError> {
        if !catalog.env.is_empty() || catalog.tools.iter().any(|tool| !tool.env.is_empty()) {
            return Err(ProviderError::validation(
                &catalog.provider.name,
                "",
                "python_persistent_env_unsupported",
                "Persistent Python providers cannot declare runtime environment requirements",
            ));
        }
        let source = std::fs::read(&path).map_err(|error| {
            ProviderError::execution(&catalog.provider.name, "", error)
                .with_phase("persistent-preflight")
        })?;
        let source_digest = sha256_hex(&source);
        let worker_group = immutable_generation_digest.unwrap_or_else(|| source_digest.clone());
        if worker_group.len() != 64 || !worker_group.as_bytes().iter().all(u8::is_ascii_hexdigit) {
            return Err(ProviderError::validation(
                &catalog.provider.name,
                "",
                "python_generation_digest_invalid",
                "immutable Python generation digest must contain exactly 64 ASCII hexadecimal characters",
            ));
        }
        let catalog_fingerprint = python_catalog_fingerprint(&catalog)
            .map_err(|error| ProviderError::execution(&catalog.provider.name, "", error))?;
        let generation_id = format!("{}-{}", catalog.provider.name, &worker_group[..16]);
        let supervisor = PythonWorkerSupervisor::new_with_capabilities(
            PythonWorkerIdentity {
                path: path.clone(),
                generation_id,
                worker_group,
                source_digest,
                catalog_fingerprint,
            },
            interpreter.clone(),
            config,
            &catalog.capabilities,
        );
        Ok(Self {
            path,
            catalog,
            env_prefix: env_prefix.into(),
            interpreter,
            supervisor: Some(supervisor),
        })
    }

    pub fn arc_persistent(
        path: PathBuf,
        catalog: ProviderCatalog,
        env_prefix: impl Into<String>,
        interpreter: PythonInterpreter,
        config: PythonSupervisorConfig,
    ) -> Result<Arc<Self>, ProviderError> {
        Self::new_persistent(path, catalog, env_prefix, interpreter, config).map(Arc::new)
    }

    /// Builds a persistent provider whose identity and worker budget are bound
    /// to the complete immutable source-tree generation.
    pub fn arc_persistent_in_generation(
        path: PathBuf,
        catalog: ProviderCatalog,
        env_prefix: impl Into<String>,
        interpreter: PythonInterpreter,
        config: PythonSupervisorConfig,
        immutable_generation_digest: String,
    ) -> Result<Arc<Self>, ProviderError> {
        Self::new_persistent_inner(
            path,
            catalog,
            env_prefix,
            interpreter,
            config,
            Some(immutable_generation_digest),
        )
        .map(Arc::new)
    }

    pub fn arc(
        path: PathBuf,
        catalog: ProviderCatalog,
        env_prefix: impl Into<String>,
    ) -> Arc<Self> {
        Arc::new(Self::new(path, catalog, env_prefix))
    }

    pub fn arc_with_interpreter(
        path: PathBuf,
        catalog: ProviderCatalog,
        env_prefix: impl Into<String>,
        interpreter: PythonInterpreter,
    ) -> Arc<Self> {
        Arc::new(Self::new_with_interpreter(
            path,
            catalog,
            env_prefix,
            interpreter,
        ))
    }
}

fn python_catalog_fingerprint(catalog: &ProviderCatalog) -> Result<String, serde_json::Error> {
    let mut catalog = catalog.clone();
    // The file source owns this location and rewrites it from the immutable
    // execution snapshot to the operator-facing provider path.
    catalog.provider.source = None;
    serde_json::to_vec(&catalog).map(|bytes| sha256_hex(&bytes))
}

fn sha256_hex(bytes: &[u8]) -> String {
    use sha2::{Digest, Sha256};
    Sha256::digest(bytes)
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect()
}

#[async_trait]
impl Provider for PythonProvider {
    fn catalog(&self) -> ProviderCatalog {
        self.catalog.clone()
    }

    async fn call(&self, call: ProviderCall) -> Result<ProviderOutput, ProviderError> {
        let tool = self.tool(&call)?;
        let runtime = PythonRuntime::from_tool(
            &self.catalog,
            tool,
            &call,
            &self.env_prefix,
            &self.interpreter,
        )?;
        let source = self.path.display().to_string();
        let env_keys = runtime.env.iter().map(|(key, _)| key.clone()).collect();
        let request = PythonWorkerRequest::call(&self.path, &call, env_keys);
        let input = encode_python_request(&request).map_err(|error| {
            ProviderError::execution(&self.catalog.provider.name, "", error)
                .with_provider_kind(self.catalog.provider.kind.as_str())
                .with_source(source.clone())
                .with_phase("input-serialization")
        })?;

        if input.len() > runtime.max_input_bytes {
            return Err(ProviderError::validation(
                &self.catalog.provider.name,
                &call.action,
                "python_input_too_large",
                format!(
                    "Python provider input exceeds {} bytes",
                    runtime.max_input_bytes
                ),
            )
            .with_provider_kind(self.catalog.provider.kind.as_str())
            .with_source(source)
            .with_phase("input-validation"));
        }

        if let Some(supervisor) = &self.supervisor {
            let value = supervisor
                .invoke_with_context(
                    &call.provider,
                    &call.action,
                    call.params.clone(),
                    supervisor::PythonInvocationOptions {
                        surface: call.surface,
                        snapshot_id: &call.snapshot_id,
                        timeout: Duration::from_millis(runtime.timeout_ms),
                        context: &call.context,
                    },
                )
                .await
                .map_err(|error| {
                    ProviderError::new(
                        error.code(),
                        &self.catalog.provider.name,
                        Some(call.action.clone()),
                        error.to_string(),
                        "Inspect the Python provider and persistent worker status, then retry.",
                    )
                    .with_provider_kind(self.catalog.provider.kind.as_str())
                    .with_source(source.clone())
                    .with_phase("persistent-execution")
                })?;
            let payload_size = serde_json::to_vec(&value)
                .map_err(|error| {
                    ProviderError::execution(&self.catalog.provider.name, &call.action, error)
                })?
                .len();
            if payload_size > runtime.max_output_bytes {
                return Err(ProviderError::validation(
                    &self.catalog.provider.name,
                    &call.action,
                    "python_output_too_large",
                    output_exceeded_message("result", runtime.max_output_bytes),
                ));
            }
            return Ok(ProviderOutput::json(value));
        }

        let started = TokioInstant::now();
        let sidecar = match run_bounded_sidecar(
            &runtime.command,
            &["-c", python_bridge_program()],
            runtime.env,
            &input,
            runtime.timeout_ms,
            protocol_output_limit(runtime.max_output_bytes),
        )
        .await
        {
            Ok(sidecar) => sidecar,
            Err(SidecarError::Timeout) => {
                return Err(ProviderError::new(
                    "python_provider_timeout",
                    &self.catalog.provider.name,
                    Some(call.action.clone()),
                    format!("Python provider exceeded {}ms timeout", runtime.timeout_ms),
                    "Increase tool.limits.timeout_ms or fix the Python provider handler.",
                )
                .with_provider_kind(self.catalog.provider.kind.as_str())
                .with_source(source)
                .with_phase("execution"));
            }
            Err(error) => {
                return Err(ProviderError::execution(
                    &self.catalog.provider.name,
                    call.action.clone(),
                    error,
                )
                .with_provider_kind(self.catalog.provider.kind.as_str())
                .with_source(source)
                .with_phase("execution"));
            }
        };
        let output = sidecar.output;

        tracing::debug!(
            provider = %self.catalog.provider.name,
            action = %call.action,
            elapsed_ms = started.elapsed().as_millis(),
            "Python provider sidecar completed"
        );

        if sidecar.stdout_exceeded
            || sidecar.stderr_exceeded
            || output.stderr.len() > runtime.max_output_bytes
        {
            let stream = if sidecar.stdout_exceeded {
                "stdout"
            } else {
                "stderr"
            };
            return Err(ProviderError::validation(
                &self.catalog.provider.name,
                &call.action,
                "python_output_too_large",
                output_exceeded_message(stream, runtime.max_output_bytes),
            )
            .with_provider_kind(self.catalog.provider.kind.as_str())
            .with_source(source)
            .with_phase("output-validation"));
        }
        if !output.status.success() {
            let stderr = String::from_utf8_lossy(&output.stderr);
            let code = if stderr.contains("python_provider_unserializable_output") {
                "python_provider_unserializable_output"
            } else {
                "python_provider_failed"
            };
            return Err(ProviderError::new(
                code,
                &self.catalog.provider.name,
                Some(call.action),
                format!("Python provider failed: {}", redact_public(&stderr)),
                "Fix the Python provider handler and retry.",
            )
            .with_provider_kind(self.catalog.provider.kind.as_str())
            .with_source(source)
            .with_phase("execution"));
        }

        let response = decode_python_response(&output.stdout).map_err(|error| {
            let code = match &error {
                PythonProtocolError::Json(_) => "python_invalid_json_output",
                _ => "python_protocol_mismatch",
            };
            ProviderError::validation(
                &self.catalog.provider.name,
                &call.action,
                code,
                error.to_string(),
            )
            .with_provider_kind(self.catalog.provider.kind.as_str())
            .with_source(source.clone())
            .with_phase("output-validation")
        })?;
        validate_python_response(&request, &response).map_err(|error| {
            ProviderError::validation(
                &self.catalog.provider.name,
                &call.action,
                "python_protocol_mismatch",
                error.to_string(),
            )
            .with_provider_kind(self.catalog.provider.kind.as_str())
            .with_source(source.clone())
            .with_phase("output-validation")
        })?;
        let value = match response {
            PythonWorkerResponse::Call { output, .. } => output,
            PythonWorkerResponse::Catalog { .. } => {
                return Err(ProviderError::validation(
                    &self.catalog.provider.name,
                    &call.action,
                    "python_protocol_mismatch",
                    "Python worker returned a catalog response for a call request",
                )
                .with_provider_kind(self.catalog.provider.kind.as_str())
                .with_source(source)
                .with_phase("output-validation"));
            }
        };
        let payload_size = serde_json::to_vec(&value)
            .map_err(|error| {
                ProviderError::execution(&self.catalog.provider.name, &call.action, error)
                    .with_provider_kind(self.catalog.provider.kind.as_str())
                    .with_source(source.clone())
                    .with_phase("output-serialization")
            })?
            .len();
        if payload_size > runtime.max_output_bytes {
            return Err(ProviderError::validation(
                &self.catalog.provider.name,
                &call.action,
                "python_output_too_large",
                output_exceeded_message("stdout", runtime.max_output_bytes),
            )
            .with_provider_kind(self.catalog.provider.kind.as_str())
            .with_source(source)
            .with_phase("output-validation"));
        }
        Ok(ProviderOutput::json(value))
    }

    fn runtime_status(&self) -> Option<Value> {
        self.supervisor
            .as_ref()
            .and_then(|supervisor| serde_json::to_value(supervisor.status()).ok())
    }

    fn cancel_active(&self) -> bool {
        self.supervisor
            .as_ref()
            .is_some_and(|supervisor| supervisor.cancel_active())
    }

    async fn reset_quarantine(&self) {
        if let Some(supervisor) = &self.supervisor {
            supervisor.reset_quarantine().await;
        }
    }

    async fn suspend(&self) {
        if let Some(supervisor) = &self.supervisor {
            supervisor.suspend().await;
        }
    }

    fn deactivate(&self) {
        if let Some(supervisor) = &self.supervisor {
            supervisor.deactivate();
        }
    }

    fn activate(&self) {
        if let Some(supervisor) = &self.supervisor {
            supervisor.activate();
        }
    }

    fn acquire_dispatch(&self) -> bool {
        self.supervisor
            .as_ref()
            .is_none_or(|supervisor| supervisor.acquire_dispatch())
    }

    fn release_dispatch(&self) {
        if let Some(supervisor) = &self.supervisor {
            supervisor.release_dispatch();
        }
    }

    async fn retire(&self) {
        if let Some(supervisor) = &self.supervisor {
            supervisor.drain_and_shutdown().await;
        }
    }
}

impl PythonProvider {
    fn tool(&self, call: &ProviderCall) -> Result<&ProviderTool, ProviderError> {
        self.catalog
            .tools
            .iter()
            .find(|tool| tool.name == call.action)
            .ok_or_else(|| {
                ProviderError::validation(
                    &self.catalog.provider.name,
                    &call.action,
                    "unknown_python_action",
                    format!("Python provider has no action `{}`", call.action),
                )
            })
    }
}

/// Introspects a `.py` provider file by importing it (in "catalog" mode) in
/// a bounded sidecar and validating the resulting manifest against the
/// generic provider-core contract. Callers that layer additional product
/// policy on top of every provider's catalog (e.g. Soma's reserved
/// CLI-command / env-prefix checks) apply it to the returned catalog
/// themselves — see the module docs above.
pub fn load_python_catalog(path: &Path, env_prefix: &str) -> Result<ProviderCatalog, String> {
    load_python_catalog_with_interpreter(path, env_prefix, &PythonInterpreter::Ambient)
}

pub fn load_python_catalog_with_interpreter(
    path: &Path,
    env_prefix: &str,
    interpreter: &PythonInterpreter,
) -> Result<ProviderCatalog, String> {
    let runtime = PythonRuntime::for_catalog(env_prefix, interpreter);
    let request = PythonWorkerRequest::catalog(path);
    let input = encode_python_request(&request).map_err(|error| error.to_string())?;
    let output = run_catalog_sidecar(&runtime, &input)?;
    let response = decode_python_response(&output).map_err(|error| error.to_string())?;
    validate_python_response(&request, &response).map_err(|error| error.to_string())?;
    let value = match response {
        PythonWorkerResponse::Catalog { catalog, .. } => catalog,
        PythonWorkerResponse::Call { .. } => {
            return Err("Python worker returned a call response for a catalog request".to_owned());
        }
    };
    let payload_size = serde_json::to_vec(&value)
        .map_err(|error| error.to_string())?
        .len();
    if payload_size > runtime.max_output_bytes {
        return Err(format!(
            "Python provider catalog {}",
            output_exceeded_message("stdout", runtime.max_output_bytes)
        ));
    }
    validate_provider_manifest_value(&value).map_err(|error| error.to_string())
}

struct PythonRuntime {
    command: String,
    env: Vec<(String, String)>,
    timeout_ms: u64,
    max_input_bytes: usize,
    max_output_bytes: usize,
}

impl PythonRuntime {
    fn for_catalog(env_prefix: &str, interpreter: &PythonInterpreter) -> Self {
        let prefix = env_prefix.trim_matches('_').to_ascii_uppercase();
        let timeout_var = format!("{prefix}_PYTHON_CATALOG_TIMEOUT_MS");
        let command_var = format!("{prefix}_PYTHON_COMMAND");
        let timeout_ms = match std::env::var(&timeout_var) {
            Ok(value) => value.parse().unwrap_or_else(|error| {
                tracing::warn!(
                    variable = %timeout_var,
                    value,
                    error = %error,
                    "invalid provider catalog timeout env var; falling back to the default"
                );
                DEFAULT_TIMEOUT_MS
            }),
            Err(_) => DEFAULT_TIMEOUT_MS,
        };
        Self {
            command: select_python_command(None, std::env::var(command_var).ok(), interpreter),
            env: Vec::new(),
            timeout_ms,
            max_input_bytes: DEFAULT_MAX_INPUT_BYTES,
            max_output_bytes: DEFAULT_MAX_OUTPUT_BYTES,
        }
    }

    fn from_tool(
        catalog: &ProviderCatalog,
        tool: &ProviderTool,
        call: &ProviderCall,
        env_prefix: &str,
        interpreter: &PythonInterpreter,
    ) -> Result<Self, ProviderError> {
        let provider_meta = catalog.meta.get("python");
        let tool_meta = tool.meta.get("python");
        let meta_field = |key: &str| {
            tool_meta
                .and_then(|value| value.get(key))
                .or_else(|| provider_meta.and_then(|value| value.get(key)))
        };
        let command = select_python_command(
            meta_field("command").and_then(Value::as_str),
            std::env::var(format!(
                "{}_PYTHON_COMMAND",
                env_prefix.trim_matches('_').to_ascii_uppercase()
            ))
            .ok(),
            interpreter,
        );
        let timeout_ms = tool
            .limits
            .as_ref()
            .and_then(|limits| limits.timeout_ms)
            .or_else(|| meta_field("timeout_ms").and_then(Value::as_u64))
            .unwrap_or(DEFAULT_TIMEOUT_MS);
        let max_input_bytes = tool
            .limits
            .as_ref()
            .and_then(|limits| limits.max_input_bytes)
            .unwrap_or(DEFAULT_MAX_INPUT_BYTES);
        let max_output_bytes = tool
            .limits
            .as_ref()
            .and_then(|limits| limits.max_response_bytes)
            .unwrap_or(DEFAULT_MAX_OUTPUT_BYTES);
        Ok(Self {
            command,
            env: collect_provider_env(
                &catalog.env,
                &tool.env,
                env_prefix,
                &call.provider,
                &call.action,
            )?,
            timeout_ms,
            max_input_bytes,
            max_output_bytes,
        })
    }
}

fn run_catalog_sidecar(runtime: &PythonRuntime, input: &[u8]) -> Result<Vec<u8>, String> {
    let mut command = StdCommand::new(resolve_sidecar_command(&runtime.command));
    command
        .args(["-c", python_bridge_program()])
        .env_clear()
        .stdin(StdStdio::piped())
        .stdout(StdStdio::piped())
        .stderr(StdStdio::piped());
    for (key, value) in sidecar_base_env() {
        command.env(key, value);
    }
    let mut child =
        spawn_retrying_busy_image_blocking(&mut command).map_err(|error| error.to_string())?;
    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| "Python provider catalog stdout pipe was not captured".to_owned())?;
    let stderr = child
        .stderr
        .take()
        .ok_or_else(|| "Python provider catalog stderr pipe was not captured".to_owned())?;
    let capture_limit = protocol_output_limit(runtime.max_output_bytes);
    let stdout_task = thread::spawn(move || read_bounded_sync(stdout, capture_limit));
    let stderr_task = thread::spawn(move || read_bounded_sync(stderr, capture_limit));

    if let Some(mut stdin) = child.stdin.take() {
        stdin.write_all(input).map_err(|error| error.to_string())?;
    }
    let deadline = Instant::now() + Duration::from_millis(runtime.timeout_ms);
    loop {
        if let Some(status) = child.try_wait().map_err(|error| error.to_string())? {
            let (stdout, stdout_exceeded) = stdout_task
                .join()
                .map_err(|_| "Python provider catalog stdout reader panicked".to_owned())?
                .map_err(|error| error.to_string())?;
            let (stderr, stderr_exceeded) = stderr_task
                .join()
                .map_err(|_| "Python provider catalog stderr reader panicked".to_owned())?
                .map_err(|error| error.to_string())?;
            if stdout_exceeded || stderr_exceeded || stderr.len() > runtime.max_output_bytes {
                let stream = if stdout_exceeded { "stdout" } else { "stderr" };
                return Err(format!(
                    "Python provider catalog {}",
                    output_exceeded_message(stream, runtime.max_output_bytes)
                ));
            }
            if !status.success() {
                return Err(format!(
                    "Python provider catalog failed: {}",
                    redact_public(&String::from_utf8_lossy(&stderr))
                ));
            }
            return Ok(stdout);
        }
        if Instant::now() >= deadline {
            let _ = child.kill();
            let _ = child.wait();
            return Err(format!(
                "Python provider catalog exceeded {}ms timeout",
                runtime.timeout_ms
            ));
        }
        std::thread::sleep(Duration::from_millis(10));
    }
}

fn read_bounded_sync<R: Read>(
    mut reader: R,
    max_output_bytes: usize,
) -> std::io::Result<(Vec<u8>, bool)> {
    let mut bytes = Vec::new();
    let mut exceeded = false;
    let mut chunk = [0u8; 8192];
    loop {
        let read = reader.read(&mut chunk)?;
        if read == 0 {
            return Ok((bytes, exceeded));
        }
        let remaining = max_output_bytes.saturating_sub(bytes.len());
        if remaining >= read && !exceeded {
            bytes.extend_from_slice(&chunk[..read]);
        } else {
            exceeded = true;
            if remaining > 0 {
                bytes.extend_from_slice(&chunk[..remaining]);
            }
        }
    }
}

#[cfg(test)]
#[path = "python_tests.rs"]
mod tests;