fix(ci): repair overhaul integration gates

Author: logannye

.github/workflows/ci.yml

@@ -113,9 +113,8 @@ jobs:
       - name: Build the rosalind binary
         run: cargo build --release --bin rosalind
       - name: Exercise the Python boundary (bit-reproducible feature substrate)
-        # Runs the documented entry point (python/rosalind.py) end-to-end: builds a
-        # toy index + BAM, extracts features twice, and proves the inputs are
-        # byte-identical with matching BLAKE3 receipts.
+        # Builds a toy reference pack + BAM, extracts native Arrow twice, and
+        # proves the streams are byte-identical with matching BLAKE3 receipts.
         run: python3 examples/reproducible_features_demo.py target/release/rosalind
 
   cli-e2e:

.github/workflows/container.yml

@@ -1,7 +1,6 @@
 name: Rosalind OCI image
 
 on:
-  workflow_dispatch:
   release:
     types: [published]
 

build.rs

@@ -1,3 +1,9 @@
 fn main() {
+    // These files are compiled into the Receipt Studio binary. Explicit change
+    // tracking prevents a restored Cargo cache from serving stale verifier
+    // assets when only the generated web bundle changed.
+    println!("cargo:rerun-if-changed=web/verify/index.html");
+    println!("cargo:rerun-if-changed=web/verify/pkg/rosalind_verify.js");
+    println!("cargo:rerun-if-changed=web/verify/pkg/rosalind_verify_bg.wasm");
     rosalind_build_info::emit();
 }

examples/reproducible_features_demo.py

@@ -1,10 +1,10 @@
 #!/usr/bin/env python3
-"""Reproducible-features demo: train a model on `rosalind features` output and
+"""Reproducible-features demo: train a model on native Arrow feature output and
 prove the inputs (and therefore the trained model) are bit-reproducible.
 
 The headline is REPRODUCIBILITY, not biological novelty: identical inputs ->
-byte-identical features -> bit-identical trained weights, verifiable by the
-BLAKE3 receipt. Runs with numpy only (no pandas/sklearn/pyarrow needed).
+byte-identical Arrow streams -> bit-identical trained weights, verifiable by the
+BLAKE3 receipt. Runs with NumPy and PyArrow (no pandas or sklearn needed).
 
     python examples/reproducible_features_demo.py [path-to-rosalind-binary]
 """
@@ -20,10 +20,16 @@
 import tempfile
 
 import numpy as np
+import pyarrow as pa
 
 REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-sys.path.insert(0, os.path.join(REPO, "python"))
-from rosalind import features  # noqa: E402
+
+NUMERIC_FEATURES = (
+    "depth", "raw_depth", "a", "c", "g", "t",
+    "a_fwd", "a_rev", "c_fwd", "c_rev",
+    "g_fwd", "g_rev", "t_fwd", "t_rev",
+    "mean_bq", "mean_mapq",
+)
 
 
 def find_binary(argv: list[str]) -> str:
@@ -44,12 +50,37 @@ def sh(cmd: list[str]) -> None:
 
 
 def output_blake3(manifest_path: str) -> str:
-    """Read the recorded BLAKE3 of the feature TSV from the run receipt."""
+    """Read the recorded BLAKE3 of the Arrow stream from the run receipt."""
     with open(manifest_path) as fh:
         m = json.load(fh)
     return m["outputs"][0]["blake3"]
 
 
+def extract_features(binary: str, reference: str, bam: str, directory: str):
+    """Write one native Arrow artifact and return its table and evidence paths."""
+    os.makedirs(directory, exist_ok=True)
+    artifact = os.path.join(directory, "features.arrow")
+    sh([
+        binary, "features", "--reference-pack", reference,
+        "--alignments", bam, "--format", "arrow-ipc", "--output", artifact,
+    ])
+    with pa.memory_map(artifact, "r") as source:
+        table = pa.ipc.open_stream(source).read_all()
+    return table, artifact, f"{artifact}.manifest.json"
+
+
+def model_arrays(table: pa.Table):
+    """Convert only the explicit training columns after bounded extraction."""
+    columns = [
+        table[name].to_numpy(zero_copy_only=False).astype(np.float64, copy=False)
+        for name in NUMERIC_FEATURES
+    ]
+    values = np.column_stack(columns)
+    positions = table["pos"].to_numpy(zero_copy_only=False)
+    references = np.asarray(table["ref"].to_pylist(), dtype="U1")
+    return values, positions, references
+
+
 def train_logreg(x: np.ndarray, y: np.ndarray, iters: int = 300, lr: float = 0.2):
     """Deterministic pure-numpy logistic regression (zero init, fixed schedule)."""
     mu = x.mean(0)
@@ -78,54 +109,61 @@ def main() -> int:
     work = tempfile.mkdtemp(prefix="rosalind-demo-")
     data = os.path.join(work, "toy")
 
-    print("== building a toy dataset + index ==")
+    print("== building a toy dataset + analysis reference pack ==")
     sh([sys.executable, os.path.join(REPO, "scripts", "generate_toy_data.py"), data])
     ref = os.path.join(data, "reference.fa")
     reads = os.path.join(data, "reads_R1.fastq")
     raw = os.path.join(work, "raw.bam")
     bam = os.path.join(work, "sorted.bam")
-    idx = os.path.join(work, "ref.idx")
+    reference_pack = os.path.join(work, "ref.rref")
     sh([binary, "align", "--reference", ref, "--reads", reads, "--format", "bam", "--output", raw])
     sh([binary, "sort", "--input", raw, "--output", bam])
-    sh([binary, "index", "--reference", ref, "--output", idx])
-
-    print("== extracting features TWICE (independent runs) ==")
-    a = features(idx, bam, binary=binary, workdir=os.path.join(work, "a"))
-    b = features(idx, bam, binary=binary, workdir=os.path.join(work, "b"))
+    sh([binary, "reference", "build", "--fasta", ref, "--output", reference_pack])
+
+    print("== extracting native Arrow features TWICE (independent runs) ==")
+    a, arrow_a, manifest_a = extract_features(
+        binary, reference_pack, bam, os.path.join(work, "a")
+    )
+    b, arrow_b, manifest_b = extract_features(
+        binary, reference_pack, bam, os.path.join(work, "b")
+    )
+    data_a, pos_a, ref_a = model_arrays(a)
+    data_b, pos_b, ref_b = model_arrays(b)
 
     # --- Reproducibility proof -------------------------------------------------
-    tsv_a = os.path.join(work, "a", "features.tsv")
-    tsv_b = os.path.join(work, "b", "features.tsv")
-    bytes_identical = open(tsv_a, "rb").read() == open(tsv_b, "rb").read()
-    h_a, h_b = output_blake3(a.manifest_path), output_blake3(b.manifest_path)
-    local_hash = hashlib.sha256(open(tsv_a, "rb").read()).hexdigest()[:16]
+    with open(arrow_a, "rb") as first, open(arrow_b, "rb") as second:
+        bytes_a, bytes_b = first.read(), second.read()
+    bytes_identical = bytes_a == bytes_b
+    h_a, h_b = output_blake3(manifest_a), output_blake3(manifest_b)
+    local_hash = hashlib.sha256(bytes_a).hexdigest()[:16]
 
-    print(f"  rows: {len(a)};  numeric features: {a.data.shape[1]}")
+    print(f"  rows: {a.num_rows};  numeric features: {data_a.shape[1]}")
     print(f"  features byte-identical across runs: {bytes_identical}")
     print(f"  receipt BLAKE3 (run A): {h_a}")
     print(f"  receipt BLAKE3 (run B): {h_b}")
     print(f"  receipts match: {h_a == h_b}")
-    assert bytes_identical, "feature TSVs differ across runs"
+    assert bytes_identical, "native Arrow streams differ across runs"
     assert h_a == h_b, "receipt hashes differ across runs"
 
     # --- A genuine supervised task: is the reference base a purine (A/G)? -------
     # Read counts peak at the ref base, so this is learnable from the features.
-    def label(ft) -> np.ndarray:
-        return np.isin(ft.ref, [b"A", b"G"]).astype(np.float64)
+    def label(references: np.ndarray) -> np.ndarray:
+        return np.isin(references, ["A", "G"]).astype(np.float64)
 
     # Deterministic train/test split by position parity.
-    def split(ft):
-        test = (ft.pos % 2) == 1
+    def split(positions: np.ndarray):
+        test = (positions % 2) == 1
         return ~test, test
 
-    ya, yb = label(a), label(b)
-    tr_a, te_a = split(a)
-    w_a, mu_a, sd_a = train_logreg(a.data[tr_a], ya[tr_a])
-    acc_a = float((predict(w_a, mu_a, sd_a, a.data[te_a]) == (ya[te_a] >= 0.5)).mean())
+    ya, yb = label(ref_a), label(ref_b)
+    tr_a, te_a = split(pos_a)
+    w_a, mu_a, sd_a = train_logreg(data_a[tr_a], ya[tr_a])
+    acc_a = float((predict(w_a, mu_a, sd_a, data_a[te_a]) == (ya[te_a] >= 0.5)).mean())
 
-    w_b, mu_b, sd_b = train_logreg(b.data[split(b)[0]], yb[split(b)[0]])
-    pred_a = predict(w_a, mu_a, sd_a, a.data[te_a])
-    pred_b = predict(w_b, mu_b, sd_b, b.data[split(b)[1]])
+    tr_b, te_b = split(pos_b)
+    w_b, mu_b, sd_b = train_logreg(data_b[tr_b], yb[tr_b])
+    pred_a = predict(w_a, mu_a, sd_a, data_a[te_a])
+    pred_b = predict(w_b, mu_b, sd_b, data_b[te_b])
 
     weights_identical = np.array_equal(w_a, w_b)
     preds_identical = np.array_equal(pred_a, pred_b)

scripts/verify-release-automation.py

@@ -54,8 +54,9 @@ def fail(message: str) -> None:
 
 for path in sorted(WORKFLOWS.glob("*.yml")):
     text = path.read_text()
-    if "packages: write" in text and path.name != "happy-image.yml":
-        fail(f"{path}: packages:write is reserved for happy-image.yml")
+    image_publishers = {"container.yml", "happy-image.yml"}
+    if "packages: write" in text and path.name not in image_publishers:
+        fail(f"{path}: packages:write is reserved for controlled image workflows")
     if "CARGO_REGISTRY_TOKEN" in text and path.name != "release.yml":
         fail(f"{path}: crates.io token is reserved for release.yml")
     if "workflow_dispatch:" in text and any(

src/genomics/sort.rs

@@ -31,7 +31,7 @@ pub fn sort_bam_deterministic(
 
     let mut reader = bam::Reader::from_path(input)
         .with_context(|| format!("failed to open BAM {}", input.display()))?;
-    let header = bam::Header::from_template(reader.header());
+    let header = coordinate_sorted_header(reader.header())?;
 
     let temp_dir = PathBuf::from(format!("{}.sort_tmp", output.display()));
     fs::create_dir_all(&temp_dir)
@@ -71,6 +71,69 @@ pub fn sort_bam_deterministic(
     Ok(())
 }
 
+/// Preserve the input dictionary and metadata while declaring the property this
+/// function has just established. `doctor` intentionally trusts the header for
+/// its cheap preflight; callers can request a full record scan separately.
+fn coordinate_sorted_header(view: &bam::HeaderView) -> Result<bam::Header> {
+    let text = std::str::from_utf8(view.as_bytes()).context("BAM header is not UTF-8")?;
+    let lines = text
+        .lines()
+        .filter(|line| !line.is_empty())
+        .collect::<Vec<_>>();
+    let mut header = bam::Header::new();
+
+    let mut hd = bam::header::HeaderRecord::new(b"HD");
+    let mut has_version = false;
+    if let Some(line) = lines
+        .iter()
+        .find(|line| line.starts_with("@HD\t") || **line == "@HD")
+    {
+        for field in line.split('\t').skip(1) {
+            let (tag, value) = field
+                .split_once(':')
+                .ok_or_else(|| anyhow!("malformed BAM @HD field {field:?}"))?;
+            if tag == "VN" {
+                has_version = true;
+            }
+            hd.push_tag(
+                tag.as_bytes(),
+                if tag == "SO" { "coordinate" } else { value },
+            );
+        }
+    }
+    if !has_version {
+        hd.push_tag(b"VN", "1.6");
+    }
+    if !text.lines().any(|line| {
+        line.starts_with("@HD\t") && line.split('\t').any(|field| field.starts_with("SO:"))
+    }) {
+        hd.push_tag(b"SO", "coordinate");
+    }
+    header.push_record(&hd);
+
+    for line in lines.into_iter().filter(|line| !line.starts_with("@HD")) {
+        if let Some(comment) = line.strip_prefix("@CO") {
+            header.push_comment(comment.strip_prefix('\t').unwrap_or(comment).as_bytes());
+            continue;
+        }
+        let mut fields = line.split('\t');
+        let kind = fields
+            .next()
+            .and_then(|value| value.strip_prefix('@'))
+            .filter(|value| value.len() == 2)
+            .ok_or_else(|| anyhow!("malformed BAM header line {line:?}"))?;
+        let mut record = bam::header::HeaderRecord::new(kind.as_bytes());
+        for field in fields {
+            let (tag, value) = field
+                .split_once(':')
+                .ok_or_else(|| anyhow!("malformed BAM header field {field:?}"))?;
+            record.push_tag(tag.as_bytes(), value);
+        }
+        header.push_record(&record);
+    }
+    Ok(header)
+}
+
 fn spill_chunk(
     header: &bam::Header,
     dir: &Path,

tests/bam_sort.rs

@@ -76,6 +76,15 @@ fn deterministic_bam_sort_orders_by_tid_pos_strand_qname() {
 
     // Read sorted output and verify order.
     let mut reader = bam::Reader::from_path(&output).unwrap();
+    let header_text = String::from_utf8_lossy(reader.header().as_bytes());
+    assert!(
+        header_text
+            .lines()
+            .next()
+            .unwrap()
+            .contains("SO:coordinate"),
+        "the sorted output must declare coordinate order: {header_text}"
+    );
     let mut qnames = Vec::new();
     let mut keys = Vec::new();
     for rec in reader.records() {

tests/doctor.rs

@@ -90,7 +90,7 @@ fn doctor_proves_a_ready_fixture_and_reports_output_safety() {
     let json = String::from_utf8(output.stdout).unwrap();
     for expected in [
         "\"ok\":true",
-        "\"declared_sort_order\":\"unknown\"",
+        "\"declared_sort_order\":\"coordinate\"",
         "\"coordinate_order_proven\":true",
         "\"output_safe\":true",
         "\"budget_feasible\":true",