feat: add the pixi_conda_script crate with the block parser and TOML model

The first piece of the `conda-script` proposal (#3751): a parser for the
`/// conda-script` comment block. The envelope parser detects the comment
prefix, extracts the TOML and maps diagnostics back into the original file;
the model covers `channels`, `entrypoint` (string or platform table),
`[dependencies]` restricted to the keys the spec defines, and `[tool.pixi]`
with `dependencies` and `pypi-dependencies`.

Author: Hofer-Julian

Cargo.lock

@@ -0,0 +1,27 @@
+[package]
+name = "pixi_conda_script"
+version = "0.1.0"
+authors.workspace = true
+edition.workspace = true
+description = "Parser for the `conda-script` metadata block embedded in code files"
+readme.workspace = true
+homepage.workspace = true
+repository.workspace = true
+license.workspace = true
+
+[dependencies]
+fs-err = { workspace = true }
+indexmap = { workspace = true }
+itertools = { workspace = true }
+miette = { workspace = true }
+pixi_pypi_spec = { workspace = true }
+pixi_spec = { workspace = true }
+pixi_toml = { workspace = true }
+rattler_conda_types = { workspace = true }
+thiserror = { workspace = true }
+toml-span = { workspace = true }
+
+[dev-dependencies]
+insta = { workspace = true }
+pixi_test_utils = { workspace = true }
+tempfile = { workspace = true }

crates/pixi_conda_script/Cargo.toml

@@ -0,0 +1,173 @@
+use std::str::FromStr;
+
+use itertools::Itertools;
+use pixi_toml::custom_error;
+use rattler_conda_types::Platform;
+use toml_span::{DeserError, Value, de_helpers::expected, value::ValueInner};
+
+/// The command that runs a `conda-script` file.
+#[derive(Debug, Clone)]
+pub enum Entrypoint {
+    /// One command for every platform.
+    Uniform(String),
+    /// A command per platform selector.
+    PerPlatform(Vec<(EntrypointSelector, String)>),
+}
+
+/// A platform key of an entrypoint table.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum EntrypointSelector {
+    /// Any Unix platform.
+    Unix,
+    /// Any Linux platform.
+    Linux,
+    /// Any macOS platform.
+    Osx,
+    /// Any Windows platform.
+    Win,
+    /// One specific conda platform.
+    Platform(Platform),
+}
+
+impl Entrypoint {
+    /// The command for `platform`, taking the most specific matching key:
+    /// the exact platform wins over its family (`linux`, `osx`, `win`),
+    /// which wins over `unix`. Returns `None` when no key matches.
+    pub fn select(&self, platform: Platform) -> Option<&str> {
+        match self {
+            Entrypoint::Uniform(command) => Some(command),
+            Entrypoint::PerPlatform(commands) => {
+                let lookup = |selector: EntrypointSelector| {
+                    commands.iter().find_map(|(candidate, command)| {
+                        (*candidate == selector).then_some(command.as_str())
+                    })
+                };
+                lookup(EntrypointSelector::Platform(platform))
+                    .or_else(|| {
+                        let family = if platform.is_linux() {
+                            EntrypointSelector::Linux
+                        } else if platform.is_osx() {
+                            EntrypointSelector::Osx
+                        } else if platform.is_windows() {
+                            EntrypointSelector::Win
+                        } else {
+                            return None;
+                        };
+                        lookup(family)
+                    })
+                    .or_else(|| {
+                        platform
+                            .is_unix()
+                            .then(|| lookup(EntrypointSelector::Unix))
+                            .flatten()
+                    })
+            }
+        }
+    }
+}
+
+impl<'de> toml_span::Deserialize<'de> for Entrypoint {
+    fn deserialize(value: &mut Value<'de>) -> Result<Self, DeserError> {
+        let span = value.span;
+        match value.take() {
+            ValueInner::String(command) => Ok(Entrypoint::Uniform(command.into_owned())),
+            ValueInner::Table(table) => {
+                if table.is_empty() {
+                    return Err(custom_error(
+                        "the entrypoint table must contain at least one platform key",
+                        span,
+                    )
+                    .into());
+                }
+                let mut errors = DeserError { errors: Vec::new() };
+                let mut commands = Vec::new();
+                for (key, mut command) in table.into_iter().sorted_by_key(|(key, _)| key.span.start)
+                {
+                    let selector = match key.name.as_ref() {
+                        "unix" => Some(EntrypointSelector::Unix),
+                        "linux" => Some(EntrypointSelector::Linux),
+                        "osx" => Some(EntrypointSelector::Osx),
+                        "win" => Some(EntrypointSelector::Win),
+                        name => match Platform::from_str(name) {
+                            Ok(platform) => Some(EntrypointSelector::Platform(platform)),
+                            Err(_) => {
+                                errors.errors.push(custom_error(
+                                    format!(
+                                        "'{name}' is neither a platform family (`unix`, `linux`, `osx`, `win`) nor a conda platform"
+                                    ),
+                                    key.span,
+                                ));
+                                None
+                            }
+                        },
+                    };
+                    let command = match command.take() {
+                        ValueInner::String(command) => Some(command.into_owned()),
+                        inner => {
+                            errors
+                                .errors
+                                .push(expected("a string", inner, command.span));
+                            None
+                        }
+                    };
+                    if let (Some(selector), Some(command)) = (selector, command) {
+                        commands.push((selector, command));
+                    }
+                }
+                if errors.errors.is_empty() {
+                    Ok(Entrypoint::PerPlatform(commands))
+                } else {
+                    Err(errors)
+                }
+            }
+            inner => Err(expected("a string or a table of platforms", inner, span).into()),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use toml_span::de_helpers::TableHelper;
+
+    use super::*;
+
+    fn parse_entrypoint(toml: &str) -> Entrypoint {
+        let mut value = toml_span::parse(toml).unwrap();
+        let mut th = TableHelper::new(&mut value).unwrap();
+        let entrypoint = th.required::<Entrypoint>("entrypoint").unwrap();
+        th.finalize(None).unwrap();
+        entrypoint
+    }
+
+    #[test]
+    fn a_uniform_entrypoint_matches_every_platform() {
+        let entrypoint = parse_entrypoint(r#"entrypoint = "python ${SCRIPT}""#);
+        assert_eq!(
+            entrypoint.select(Platform::Linux64),
+            Some("python ${SCRIPT}")
+        );
+        assert_eq!(entrypoint.select(Platform::Win64), Some("python ${SCRIPT}"));
+    }
+
+    #[test]
+    fn the_most_specific_platform_key_wins() {
+        let entrypoint = parse_entrypoint(
+            r#"entrypoint = { unix = "unix", linux = "linux", linux-64 = "linux-64", win = "win" }"#,
+        );
+        assert_eq!(entrypoint.select(Platform::Linux64), Some("linux-64"));
+        assert_eq!(entrypoint.select(Platform::LinuxAarch64), Some("linux"));
+        assert_eq!(entrypoint.select(Platform::Osx64), Some("unix"));
+        assert_eq!(entrypoint.select(Platform::Win64), Some("win"));
+        assert_eq!(entrypoint.select(Platform::WinArm64), Some("win"));
+    }
+
+    #[test]
+    fn a_platform_without_a_matching_key_selects_nothing() {
+        let windows_only = parse_entrypoint(r#"entrypoint = { win = "win" }"#);
+        assert_eq!(windows_only.select(Platform::Linux64), None);
+
+        let unix_only = parse_entrypoint(r#"entrypoint = { unix = "unix" }"#);
+        assert_eq!(unix_only.select(Platform::Win64), None);
+        assert_eq!(unix_only.select(Platform::LinuxRiscv64), Some("unix"));
+    }
+}

crates/pixi_conda_script/src/entrypoint.rs

@@ -0,0 +1,168 @@
+use std::ops::Range;
+
+use miette::SourceSpan;
+
+use crate::error::EnvelopeErrorKind;
+
+pub(crate) const OPENING_MARKER: &str = "/// conda-script";
+pub(crate) const CLOSING_MARKER: &str = "/// end-conda-script";
+const PEP723_OPENING: &str = "# /// script";
+
+/// The extracted content of a `conda-script` block.
+pub(crate) struct CondaScriptBlock {
+    /// The block content with the comment prefix stripped from every line.
+    pub(crate) metadata: String,
+    pub(crate) source_map: SourceMap,
+}
+
+/// Maps offsets in the extracted metadata TOML back to the original file.
+#[derive(Debug, Clone)]
+pub(crate) struct SourceMap {
+    opening: Range<usize>,
+    metadata_lines: Vec<MetadataLine>,
+}
+
+#[derive(Debug, Clone)]
+struct MetadataLine {
+    metadata_start: usize,
+    source_start: usize,
+    len: usize,
+}
+
+impl SourceMap {
+    fn metadata_offset(&self, offset: usize) -> usize {
+        let Some(line) = self
+            .metadata_lines
+            .iter()
+            .rev()
+            .find(|line| line.metadata_start <= offset)
+        else {
+            return self.opening.start;
+        };
+        line.source_start + offset.saturating_sub(line.metadata_start).min(line.len)
+    }
+
+    pub(crate) fn span(&self, offset: usize, len: usize) -> SourceSpan {
+        let start = self.metadata_offset(offset);
+        let end = self.metadata_offset(offset.saturating_add(len)).max(start);
+        SourceSpan::new(start.into(), end - start)
+    }
+}
+
+/// Extracts the single `conda-script` block from a source file.
+///
+/// Returns `Ok(None)` when the file contains no opening marker with a valid
+/// comment prefix.
+pub(crate) fn parse_block(source: &str) -> Result<Option<CondaScriptBlock>, EnvelopeErrorKind> {
+    // A BOM must not become part of the comment prefix of a block on the
+    // first line.
+    let base = if source.starts_with('\u{feff}') {
+        '\u{feff}'.len_utf8()
+    } else {
+        0
+    };
+
+    let mut lines: Vec<(usize, &str)> = Vec::new();
+    let mut offset = base;
+    for raw_line in source[base..].split_inclusive('\n') {
+        lines.push((offset, without_line_ending(raw_line)));
+        offset += raw_line.len();
+    }
+
+    let Some((opening_index, prefix)) = lines
+        .iter()
+        .enumerate()
+        .find_map(|(index, (_, line))| opening_prefix(line).map(|prefix| (index, prefix)))
+    else {
+        return Ok(None);
+    };
+    let opening = line_span(lines[opening_index]);
+
+    let mut toml_lines: Vec<&str> = Vec::new();
+    let mut metadata_lines = Vec::new();
+    let mut metadata_len = 0;
+    let mut closing_index = None;
+    let mut broken_line = None;
+    for (index, &(line_start, line)) in lines.iter().enumerate().skip(opening_index + 1) {
+        if let Some(rest) = line.strip_prefix(prefix) {
+            if rest.trim_end() == CLOSING_MARKER {
+                closing_index = Some(index);
+                break;
+            }
+            toml_lines.push(rest);
+            metadata_lines.push(MetadataLine {
+                metadata_start: metadata_len,
+                source_start: line_start + prefix.len(),
+                len: rest.len(),
+            });
+            metadata_len += rest.len() + 1;
+        } else if line.trim_end() == prefix.trim_end() {
+            toml_lines.push("");
+            metadata_lines.push(MetadataLine {
+                metadata_start: metadata_len,
+                source_start: line_start + line.trim_end().len(),
+                len: 0,
+            });
+            metadata_len += 1;
+        } else {
+            broken_line = Some(line_span((line_start, line)));
+            break;
+        }
+    }
+
+    let Some(closing_index) = closing_index else {
+        return Err(EnvelopeErrorKind::Unterminated {
+            opening,
+            broken_line,
+            prefix: prefix.to_owned(),
+        });
+    };
+
+    if let Some(second) = lines[closing_index + 1..]
+        .iter()
+        .find(|(_, line)| opening_prefix(line).is_some())
+    {
+        return Err(EnvelopeErrorKind::MultipleBlocks {
+            first: opening,
+            second: line_span(*second),
+        });
+    }
+
+    if let Some(pep723) = lines[..opening_index]
+        .iter()
+        .chain(&lines[closing_index + 1..])
+        .find(|(_, line)| line.trim_end() == PEP723_OPENING)
+    {
+        return Err(EnvelopeErrorKind::BothBlockKinds {
+            conda_script: opening,
+            pep723: line_span(*pep723),
+        });
+    }
+
+    Ok(Some(CondaScriptBlock {
+        metadata: toml_lines.join("\n") + "\n",
+        source_map: SourceMap {
+            opening,
+            metadata_lines,
+        },
+    }))
+}
+
+fn line_span((start, line): (usize, &str)) -> Range<usize> {
+    start..start + line.trim_end().len()
+}
+
+/// The comment prefix when `line` opens a `conda-script` block.
+///
+/// A prefix must be non-empty and free of alphanumeric characters, so a
+/// mention of the marker inside code (`x = "// /// conda-script"`) does not
+/// open a block.
+fn opening_prefix(line: &str) -> Option<&str> {
+    let prefix = line.trim_end().strip_suffix(OPENING_MARKER)?;
+    (!prefix.is_empty() && !prefix.contains(char::is_alphanumeric)).then_some(prefix)
+}
+
+fn without_line_ending(line: &str) -> &str {
+    let line = line.strip_suffix('\n').unwrap_or(line);
+    line.strip_suffix('\r').unwrap_or(line)
+}