From 8d786aeecc6e7e815ab8a021bebcb5c17004ddbb Mon Sep 17 00:00:00 2001 From: Lumi AI Workspace Date: Mon, 6 Jul 2026 23:53:53 +0700 Subject: [PATCH 1/4] Make lab batch packs fail closed --- scripts/lab/build_lab_batch_pack.py | 187 ++++++++++++++++++++-------- 1 file changed, 137 insertions(+), 50 deletions(-) diff --git a/scripts/lab/build_lab_batch_pack.py b/scripts/lab/build_lab_batch_pack.py index 09c9c6dd..65661e90 100644 --- a/scripts/lab/build_lab_batch_pack.py +++ b/scripts/lab/build_lab_batch_pack.py @@ -1,7 +1,11 @@ -"""Build a zip archive with everything a lab partner needs. +"""Build a fail-closed zip archive with everything a lab partner needs. -Collects candidate CSV, evidence certificates, protocol references, -controls manifest, and data return templates into a single zip. +Collects the candidate panel, matching evidence certificates, safe review +references, controls manifest, data-return template, and chain-of-custody +hashes into a single archive. + +This module intentionally validates identity and review completeness. It does +not provide wet-lab procedures and does not assert biological activity. """ from __future__ import annotations @@ -15,6 +19,15 @@ from pathlib import Path from typing import Any +REQUIRED_PANEL_COLUMNS = {"candidate_id", "sequence"} +DEFAULT_PROTOCOL_DOCS = [ + "docs/review/LAB_PARTNER_ONBOARDING.md", + "docs/review/PRE_REGISTERED_PILOT_TEMPLATE.md", + "docs/review/EXTERNAL_REVIEW_PACKET.md", + "docs/evidence/PROOF_LADDER.md", + "docs/trust/TRUST_CENTER.md", +] + def _sha256_text(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() @@ -26,16 +39,35 @@ def _sha256_bytes(value: bytes) -> str: def _read_panel_rows(panel_csv: Path) -> list[dict[str, str]]: with panel_csv.open(newline="", encoding="utf-8") as f: - return list(csv.DictReader(f)) + reader = csv.DictReader(f) + fieldnames = set(reader.fieldnames or []) + missing = sorted(REQUIRED_PANEL_COLUMNS - fieldnames) + if missing: + raise ValueError( + "Panel CSV is missing required column(s): " + ", ".join(missing) + ) + rows = list(reader) + + if not rows: + raise ValueError("Panel CSV contains no candidate rows.") + + if not _candidate_identity_rows(rows): + raise ValueError("Panel CSV contains no valid candidate identities.") + + return rows def _candidate_identity_rows(panel_rows: list[dict[str, str]]) -> list[dict[str, Any]]: identities: list[dict[str, Any]] = [] + seen: set[str] = set() for row in panel_rows: candidate_id = row.get("candidate_id", "").strip() sequence = row.get("sequence", "").strip().upper() if not candidate_id or not sequence: continue + if candidate_id in seen: + raise ValueError(f"Duplicate candidate_id in panel CSV: {candidate_id}") + seen.add(candidate_id) identities.append({ "candidate_id": candidate_id, "sequence_sha256": _sha256_text(sequence), @@ -55,6 +87,16 @@ def _synthesis_order_digest(panel_rows: list[dict[str, str]]) -> str: return _sha256_text("\n".join(lines) + "\n") +def _protocol_doc_paths(protocol_docs: list[str] | None) -> tuple[list[Path], list[str]]: + docs = DEFAULT_PROTOCOL_DOCS if protocol_docs is None else protocol_docs + if not docs: + return [], ["at least one safe review reference document is required"] + + paths = [Path(doc) for doc in docs] + missing = [str(path) for path in paths if not path.exists()] + return paths, missing + + def verify_batch_pack(zip_path: str | Path) -> dict[str, Any]: """Verify chain-of-custody hashes inside a lab batch pack zip.""" zip_p = Path(zip_path) @@ -108,9 +150,9 @@ def build_batch_pack( controls: list[dict[str, Any]] | None = None, data_return_schema: str = "schemas/lab_result.schema.json", ) -> dict[str, Any]: - """Build a lab batch pack zip. + """Build a fail-closed lab batch pack zip. - Returns manifest dict with included files and counts. + Returns a manifest dict with included files and counts, or an ``error`` dict. """ panel_p = Path(panel_csv) evidence_p = Path(evidence_dir) @@ -122,14 +164,51 @@ def build_batch_pack( if not evidence_p.exists(): return {"error": f"Evidence directory not found: {evidence_dir}"} - panel_rows = _read_panel_rows(panel_p) + try: + panel_rows = _read_panel_rows(panel_p) + except ValueError as exc: + return {"error": str(exc)} + + protocol_paths, missing_protocol_docs = _protocol_doc_paths(protocol_docs) + if missing_protocol_docs: + return { + "error": "Missing protocol/reference document(s): " + + ", ".join(missing_protocol_docs), + "missing_protocol_docs": missing_protocol_docs, + } + candidate_identities = _candidate_identity_rows(panel_rows) + panel_ids = {row["candidate_id"] for row in candidate_identities} + + evidence_files = { + f.stem: f + for f in sorted(evidence_p.iterdir()) + if f.is_file() and f.suffix == ".json" + } + evidence_ids = set(evidence_files) + + missing_evidence = sorted(panel_ids - evidence_ids) + extra_evidence = sorted(evidence_ids - panel_ids) + if missing_evidence: + return { + "error": "Missing evidence certificate(s) for panel candidate(s): " + + ", ".join(missing_evidence), + "missing_evidence": missing_evidence, + } + if extra_evidence: + return { + "error": "Evidence directory contains certificate(s) not present in panel CSV: " + + ", ".join(extra_evidence), + "extra_evidence": extra_evidence, + } + panel_data = panel_p.read_bytes() panel_csv_sha256 = _sha256_bytes(panel_data) synthesis_order_sha256 = _synthesis_order_digest(panel_rows) manifest: dict[str, Any] = { "generated_at": datetime.now(timezone.utc).isoformat(), + "pack_standard": "fail_closed_lab_handoff_v1", "panel_csv": panel_csv, "panel_csv_sha256": panel_csv_sha256, "synthesis_order_sha256": synthesis_order_sha256, @@ -140,54 +219,53 @@ def build_batch_pack( "candidate_identities": candidate_identities, "data_return_template": None, "controls": controls or [], + "strict_identity_checks": { + "panel_schema_required": sorted(REQUIRED_PANEL_COLUMNS), + "evidence_must_match_panel_exactly": True, + "protocol_references_required": True, + }, } + out_p.parent.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(out_p, "w", zipfile.ZIP_DEFLATED) as zf: # 1. Panel CSV zf.writestr("panel.csv", panel_data) manifest["panel_csv_size"] = len(panel_data) - # 2. Evidence certificates - ev_count = 0 + # 2. Evidence certificates, exact-match to panel candidate IDs evidence_hashes: dict[str, str] = {} - for f in sorted(evidence_p.iterdir()): - if f.suffix == ".json": - archive_name = f"evidence/{f.name}" - evidence_bytes = f.read_bytes() - zf.writestr(archive_name, evidence_bytes) - evidence_hashes[archive_name] = _sha256_bytes(evidence_bytes) - ev_count += 1 - manifest["included_candidates"].append(f.stem) - manifest["evidence_count"] = ev_count + for candidate_id in sorted(panel_ids): + f = evidence_files[candidate_id] + archive_name = f"evidence/{f.name}" + evidence_bytes = f.read_bytes() + zf.writestr(archive_name, evidence_bytes) + evidence_hashes[archive_name] = _sha256_bytes(evidence_bytes) + manifest["included_candidates"].append(candidate_id) + manifest["evidence_count"] = len(evidence_hashes) manifest["evidence_certificate_sha256"] = evidence_hashes - # 3. Protocol references (as a README) - protocols = protocol_docs or [ - "docs/LAB_PARTNER_ONBOARDING.md", - "docs/ASSAY_PREREGISTRATION.md", - "docs/WET_LAB_HANDOFF.md", - ] + # 3. Safe review references. These are not operational wet-lab protocols. protocol_lines = [ - "# Lab Batch Pack — Protocol References", + "# Lab Batch Pack — Safe Review References", "", - "See the following documents for assay protocols and instructions:", + "The archive includes these review documents, not experimental instructions:", "", ] - for doc_path in protocols: - doc_p = Path(doc_path) - if doc_p.exists(): - zf.write(str(doc_p), f"protocols/{doc_p.name}") - protocol_lines.append(f"- {doc_path}") - manifest["included_protocols"].append(doc_path) - zf.writestr("protocols/README.md", "\n".join(protocol_lines)) + for doc_p in protocol_paths: + zf.write(str(doc_p), f"protocols/{doc_p.name}") + protocol_lines.append(f"- {doc_p}") + manifest["included_protocols"].append(str(doc_p)) + zf.writestr("protocols/README.md", "\n".join(protocol_lines) + "\n") # 4. Controls manifest ctrl = controls or [ - {"candidate_id": "SEED-001_VAR_064", - "role": "POSITIVE_CONTROL", - "sequence": "GIGKFLHSAKKFGKAFVGEIMNS", - "expected_mic_range_ugml": "4-32", - "notes": "Magainin-1 derivative; validates MIC assay"}, + { + "candidate_id": "SEED-001_VAR_064", + "role": "POSITIVE_CONTROL", + "sequence": "GIGKFLHSAKKFGKAFVGEIMNS", + "expected_mic_range_ugml": "4-32", + "notes": "Magainin-1 derivative; validates MIC assay", + }, ] zf.writestr("controls_manifest.json", json.dumps(ctrl, indent=2)) manifest["controls"] = ctrl @@ -214,11 +292,13 @@ def build_batch_pack( # 6. Chain-of-custody identity hashes custody = { - "chain_of_custody_version": "1.0", + "chain_of_custody_version": "1.1", "scope": ( "Dry-lab identity check only. Hashes verify that candidate IDs, " "sequences, panel ordering, and evidence certificate files did not drift." ), + "pack_standard": "fail_closed_lab_handoff_v1", + "strict_identity_checks": manifest["strict_identity_checks"], "not_evidence_of": [ "biological activity", "mammalian safety", @@ -233,33 +313,40 @@ def build_batch_pack( zf.writestr("chain_of_custody.json", json.dumps(custody, indent=2)) zf.writestr("MANIFEST.json", json.dumps(manifest, indent=2)) - # 6. Batch README + # 7. Batch README readme = [ "# OpenAMP Foundry — Lab Batch Pack", "", f"Generated: {manifest['generated_at']}", - f"Candidates: {ev_count}", - f"Protocols: {len(manifest['included_protocols'])}", + f"Candidates: {manifest['candidate_count']}", + f"Evidence certificates: {manifest['evidence_count']}", + f"Review references: {len(manifest['included_protocols'])}", "", "## Contents", "", "- panel.csv — Candidate list with IDs and sequences", - "- evidence/ — Evidence certificates (one JSON per candidate)", - "- protocols/ — Assay protocol documents", - "- controls_manifest.json — Positive/negative control sequences", + "- evidence/ — Evidence certificates, one JSON per panel candidate", + "- protocols/ — Safe review references, not wet-lab procedures", + "- controls_manifest.json — Positive/negative control metadata", "- data_return_template.json — Empty JSON template per lab_result.schema.json", - "- chain_of_custody.json — SHA-256 identity hashes for panel, sequences, and evidence files", + "- chain_of_custody.json — SHA-256 identity hashes", "- MANIFEST.json — Machine-readable archive manifest", "", + "## Fail-Closed Checks Applied", + "", + "- panel.csv must contain candidate_id and sequence columns", + "- every panel candidate must have exactly one evidence certificate", + "- evidence_dir must not contain certificates outside the panel", + "- safe review-reference documents must be bundled", + "", "## Identity Verification", "", "Before using returned results, verify that panel and evidence files match:", "", - " python scripts/build_lab_batch_pack.py --verify-pack ", - "", - "These hashes verify identity only. They are not evidence of activity, safety, synthesis success, or clinical utility.", + " python scripts/lab/build_lab_batch_pack.py --verify-pack ", "", - "## Return Data", + "These hashes verify identity only. They are not evidence of activity, safety, " + "synthesis success, or clinical utility.", "", "After assaying, return one JSON file per candidate per assay type.", "Use the data_return_template.json as a starting point.", From f7a4d49ad0f085f72beba6170850045e3619f59d Mon Sep 17 00:00:00 2001 From: Lumi AI Workspace Date: Mon, 6 Jul 2026 23:54:26 +0700 Subject: [PATCH 2/4] Test fail-closed lab batch packs --- tests/lab/test_lab_batch_pack.py | 137 ++++++++++++++++++++++++------- 1 file changed, 108 insertions(+), 29 deletions(-) diff --git a/tests/lab/test_lab_batch_pack.py b/tests/lab/test_lab_batch_pack.py index 5e2e24e3..496cba3d 100644 --- a/tests/lab/test_lab_batch_pack.py +++ b/tests/lab/test_lab_batch_pack.py @@ -1,4 +1,4 @@ -"""Tests for the lab batch pack builder.""" +"""Tests for the fail-closed lab batch pack builder.""" import importlib.util import json @@ -17,27 +17,32 @@ verify_batch_pack = _mod.verify_batch_pack -def _make_evidence(tmp: Path, candidate_id: str) -> Path: - """Create a minimal evidence certificate.""" +def _make_evidence(tmp: Path, *candidate_ids: str) -> Path: + """Create minimal evidence certificates for candidate IDs.""" d = tmp / "evidence" d.mkdir(exist_ok=True) - (d / f"{candidate_id}.json").write_text( - json.dumps({"candidate_id": candidate_id, "scores": {"ensemble": 0.8}}) - ) + for candidate_id in candidate_ids: + (d / f"{candidate_id}.json").write_text( + json.dumps({"candidate_id": candidate_id, "scores": {"ensemble": 0.8}}) + ) return d -def _make_panel_csv(tmp: Path) -> Path: +def _make_panel_csv(tmp: Path, rows: list[tuple[str, str]] | None = None) -> Path: """Create a minimal panel CSV.""" + rows = rows or [ + ("C1", "KKLFKKILKYL"), + ("C2", "ACDEFGHIKLMNPQRSTVWY"), + ] p = tmp / "panel.csv" - p.write_text("candidate_id,sequence\nC1,KKLFKKILKYL\nC2,ACDEFGHIKLMNPQRSTVWY\n") + body = "\n".join(f"{candidate_id},{sequence}" for candidate_id, sequence in rows) + p.write_text(f"candidate_id,sequence\n{body}\n") return p def test_build_pack_success(tmp_path): panel = _make_panel_csv(tmp_path) - ev_dir = _make_evidence(tmp_path, "C1") - _make_evidence(tmp_path, "C2") + ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" result = build_batch_pack( panel_csv=str(panel), @@ -46,31 +51,36 @@ def test_build_pack_success(tmp_path): ) assert "error" not in result assert out.exists() - assert result["evidence_count"] >= 2 + assert result["evidence_count"] == 2 + assert result["strict_identity_checks"]["evidence_must_match_panel_exactly"] is True def test_build_pack_contains_evidence(tmp_path): panel = _make_panel_csv(tmp_path) - ev_dir = _make_evidence(tmp_path, "C1") + ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) with zipfile.ZipFile(out) as z: names = z.namelist() - assert any("evidence/C1.json" in n for n in names) + assert "evidence/C1.json" in names + assert "evidence/C2.json" in names def test_build_pack_contains_readme(tmp_path): panel = _make_panel_csv(tmp_path) - ev_dir = _make_evidence(tmp_path, "C1") + ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) with zipfile.ZipFile(out) as z: assert "README.md" in z.namelist() + readme = z.read("README.md").decode("utf-8") + assert "Fail-Closed Checks Applied" in readme + assert "not wet-lab procedures" in readme def test_build_pack_contains_controls_manifest(tmp_path): panel = _make_panel_csv(tmp_path) - ev_dir = _make_evidence(tmp_path, "C1") + ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) with zipfile.ZipFile(out) as z: @@ -79,7 +89,7 @@ def test_build_pack_contains_controls_manifest(tmp_path): def test_build_pack_contains_data_return_template(tmp_path): panel = _make_panel_csv(tmp_path) - ev_dir = _make_evidence(tmp_path, "C1") + ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) with zipfile.ZipFile(out) as z: @@ -88,8 +98,7 @@ def test_build_pack_contains_data_return_template(tmp_path): def test_build_pack_contains_chain_of_custody(tmp_path): panel = _make_panel_csv(tmp_path) - ev_dir = _make_evidence(tmp_path, "C1") - _make_evidence(tmp_path, "C2") + ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" manifest = build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) with zipfile.ZipFile(out) as z: @@ -97,6 +106,7 @@ def test_build_pack_contains_chain_of_custody(tmp_path): assert "MANIFEST.json" in z.namelist() assert custody["panel_csv_sha256"] == manifest["panel_csv_sha256"] assert custody["synthesis_order_sha256"] == manifest["synthesis_order_sha256"] + assert custody["pack_standard"] == "fail_closed_lab_handoff_v1" assert len(custody["candidate_identities"]) == 2 assert custody["candidate_identities"][0]["candidate_id"] == "C1" assert "sequence_sha256" in custody["candidate_identities"][0] @@ -105,8 +115,7 @@ def test_build_pack_contains_chain_of_custody(tmp_path): def test_verify_batch_pack_success(tmp_path): panel = _make_panel_csv(tmp_path) - ev_dir = _make_evidence(tmp_path, "C1") - _make_evidence(tmp_path, "C2") + ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) result = verify_batch_pack(out) @@ -119,7 +128,7 @@ def test_verify_batch_pack_success(tmp_path): def test_verify_batch_pack_detects_tampered_panel(tmp_path): panel = _make_panel_csv(tmp_path) - ev_dir = _make_evidence(tmp_path, "C1") + ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) @@ -138,15 +147,68 @@ def test_verify_batch_pack_detects_tampered_panel(tmp_path): assert result["synthesis_order_hash_ok"] is False -def test_build_pack_contains_protocols(tmp_path): +def test_build_pack_contains_real_review_reference_docs(tmp_path): panel = _make_panel_csv(tmp_path) - ev_dir = _make_evidence(tmp_path, "C1") + ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" - build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) + result = build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) + assert "error" not in result with zipfile.ZipFile(out) as z: names = z.namelist() - protocol_files = [n for n in names if n.startswith("protocols/")] - assert len(protocol_files) > 0 + assert "protocols/README.md" in names + assert "protocols/LAB_PARTNER_ONBOARDING.md" in names + assert "protocols/PRE_REGISTERED_PILOT_TEMPLATE.md" in names + assert "protocols/EXTERNAL_REVIEW_PACKET.md" in names + + +def test_build_pack_fails_for_missing_required_panel_column(tmp_path): + panel = tmp_path / "panel.csv" + panel.write_text("id,seq\nC1,KKLFKKILKYL\n") + ev_dir = _make_evidence(tmp_path, "C1") + result = build_batch_pack( + panel_csv=str(panel), + evidence_dir=str(ev_dir), + out_zip=str(tmp_path / "pack.zip"), + ) + assert "error" in result + assert "missing required column" in result["error"] + + +def test_build_pack_fails_for_missing_evidence_certificate(tmp_path): + panel = _make_panel_csv(tmp_path) + ev_dir = _make_evidence(tmp_path, "C1") + result = build_batch_pack( + panel_csv=str(panel), + evidence_dir=str(ev_dir), + out_zip=str(tmp_path / "pack.zip"), + ) + assert "error" in result + assert result["missing_evidence"] == ["C2"] + + +def test_build_pack_fails_for_extra_evidence_certificate(tmp_path): + panel = _make_panel_csv(tmp_path, rows=[("C1", "KKLFKKILKYL")]) + ev_dir = _make_evidence(tmp_path, "C1", "C2") + result = build_batch_pack( + panel_csv=str(panel), + evidence_dir=str(ev_dir), + out_zip=str(tmp_path / "pack.zip"), + ) + assert "error" in result + assert result["extra_evidence"] == ["C2"] + + +def test_build_pack_fails_for_missing_protocol_doc(tmp_path): + panel = _make_panel_csv(tmp_path) + ev_dir = _make_evidence(tmp_path, "C1", "C2") + result = build_batch_pack( + panel_csv=str(panel), + evidence_dir=str(ev_dir), + out_zip=str(tmp_path / "pack.zip"), + protocol_docs=[str(tmp_path / "missing.md")], + ) + assert "error" in result + assert "Missing protocol/reference document" in result["error"] def test_build_pack_error_missing_panel(tmp_path): @@ -168,7 +230,7 @@ def test_build_pack_error_missing_evidence(tmp_path): def test_cli_runs(tmp_path): - ev_dir = _make_evidence(tmp_path, "C1") + ev_dir = _make_evidence(tmp_path, "C1", "C2") panel = _make_panel_csv(tmp_path) out = tmp_path / "pack.zip" result = subprocess.run( @@ -184,7 +246,7 @@ def test_cli_runs(tmp_path): def test_cli_manifest_output(tmp_path): - ev_dir = _make_evidence(tmp_path, "C1") + ev_dir = _make_evidence(tmp_path, "C1", "C2") panel = _make_panel_csv(tmp_path) out = tmp_path / "pack.zip" manifest = tmp_path / "manifest.json" @@ -203,12 +265,29 @@ def test_cli_manifest_output(tmp_path): assert "evidence_count" in data assert "panel_csv_sha256" in data assert "synthesis_order_sha256" in data + assert data["pack_standard"] == "fail_closed_lab_handoff_v1" -def test_cli_verify_pack(tmp_path): +def test_cli_returns_2_for_missing_evidence(tmp_path): ev_dir = _make_evidence(tmp_path, "C1") panel = _make_panel_csv(tmp_path) out = tmp_path / "pack.zip" + result = subprocess.run( + [sys.executable, "scripts/lab/build_lab_batch_pack.py", + "--panel-csv", str(panel), + "--evidence-dir", str(ev_dir), + "--out", str(out)], + capture_output=True, text=True, + env={"PYTHONPATH": "src"}, + ) + assert result.returncode == 2 + assert "Missing evidence certificate" in result.stderr + + +def test_cli_verify_pack(tmp_path): + ev_dir = _make_evidence(tmp_path, "C1", "C2") + panel = _make_panel_csv(tmp_path) + out = tmp_path / "pack.zip" build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) result = subprocess.run( [sys.executable, "scripts/lab/build_lab_batch_pack.py", "--verify-pack", str(out)], From 799690d73693e7efb81b133d950d01ea0cb32466 Mon Sep 17 00:00:00 2001 From: Lumi AI Workspace Date: Tue, 7 Jul 2026 00:07:44 +0700 Subject: [PATCH 3/4] Fix lint in lab batch pack builder --- scripts/lab/build_lab_batch_pack.py | 78 +++++++++++++++++------------ 1 file changed, 45 insertions(+), 33 deletions(-) diff --git a/scripts/lab/build_lab_batch_pack.py b/scripts/lab/build_lab_batch_pack.py index 65661e90..1c4490fa 100644 --- a/scripts/lab/build_lab_batch_pack.py +++ b/scripts/lab/build_lab_batch_pack.py @@ -37,6 +37,29 @@ def _sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() +def _candidate_identity_rows( + panel_rows: list[dict[str, str]], +) -> list[dict[str, Any]]: + identities: list[dict[str, Any]] = [] + seen: set[str] = set() + for row in panel_rows: + candidate_id = row.get("candidate_id", "").strip() + sequence = row.get("sequence", "").strip().upper() + if not candidate_id or not sequence: + continue + if candidate_id in seen: + raise ValueError(f"Duplicate candidate_id in panel CSV: {candidate_id}") + seen.add(candidate_id) + identities.append( + { + "candidate_id": candidate_id, + "sequence_sha256": _sha256_text(sequence), + "sequence_length": len(sequence), + } + ) + return identities + + def _read_panel_rows(panel_csv: Path) -> list[dict[str, str]]: with panel_csv.open(newline="", encoding="utf-8") as f: reader = csv.DictReader(f) @@ -57,25 +80,6 @@ def _read_panel_rows(panel_csv: Path) -> list[dict[str, str]]: return rows -def _candidate_identity_rows(panel_rows: list[dict[str, str]]) -> list[dict[str, Any]]: - identities: list[dict[str, Any]] = [] - seen: set[str] = set() - for row in panel_rows: - candidate_id = row.get("candidate_id", "").strip() - sequence = row.get("sequence", "").strip().upper() - if not candidate_id or not sequence: - continue - if candidate_id in seen: - raise ValueError(f"Duplicate candidate_id in panel CSV: {candidate_id}") - seen.add(candidate_id) - identities.append({ - "candidate_id": candidate_id, - "sequence_sha256": _sha256_text(sequence), - "sequence_length": len(sequence), - }) - return identities - - def _synthesis_order_digest(panel_rows: list[dict[str, str]]) -> str: """Hash the ordered candidate identity list, not vendor instructions.""" lines = [] @@ -87,7 +91,9 @@ def _synthesis_order_digest(panel_rows: list[dict[str, str]]) -> str: return _sha256_text("\n".join(lines) + "\n") -def _protocol_doc_paths(protocol_docs: list[str] | None) -> tuple[list[Path], list[str]]: +def _protocol_doc_paths( + protocol_docs: list[str] | None, +) -> tuple[list[Path], list[str]]: docs = DEFAULT_PROTOCOL_DOCS if protocol_docs is None else protocol_docs if not docs: return [], ["at least one safe review reference document is required"] @@ -114,18 +120,26 @@ def verify_batch_pack(zip_path: str | Path) -> dict[str, Any]: custody = json.loads(zf.read("chain_of_custody.json").decode("utf-8")) manifest = json.loads(zf.read("MANIFEST.json").decode("utf-8")) - panel_hash_ok = _sha256_bytes(panel_bytes) == custody.get("panel_csv_sha256") + panel_hash_ok = _sha256_bytes(panel_bytes) == custody.get( + "panel_csv_sha256" + ) panel_text = panel_bytes.decode("utf-8") panel_rows = list(csv.DictReader(panel_text.splitlines())) expected_identities = _candidate_identity_rows(panel_rows) identities_ok = expected_identities == custody.get("candidate_identities", []) - order_ok = _synthesis_order_digest(panel_rows) == custody.get("synthesis_order_sha256") + order_ok = _synthesis_order_digest(panel_rows) == custody.get( + "synthesis_order_sha256" + ) evidence_ok = True evidence_hashes = custody.get("evidence_certificate_sha256", {}) for archive_name, expected_hash in evidence_hashes.items(): - if archive_name not in names or _sha256_bytes(zf.read(archive_name)) != expected_hash: + if archive_name not in names: + evidence_ok = False + break + evidence_hash = _sha256_bytes(zf.read(archive_name)) + if evidence_hash != expected_hash: evidence_ok = False break @@ -196,9 +210,9 @@ def build_batch_pack( "missing_evidence": missing_evidence, } if extra_evidence: + error = "Evidence directory contains certificate(s) not present in panel CSV: " return { - "error": "Evidence directory contains certificate(s) not present in panel CSV: " - + ", ".join(extra_evidence), + "error": error + ", ".join(extra_evidence), "extra_evidence": extra_evidence, } @@ -228,11 +242,9 @@ def build_batch_pack( out_p.parent.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(out_p, "w", zipfile.ZIP_DEFLATED) as zf: - # 1. Panel CSV zf.writestr("panel.csv", panel_data) manifest["panel_csv_size"] = len(panel_data) - # 2. Evidence certificates, exact-match to panel candidate IDs evidence_hashes: dict[str, str] = {} for candidate_id in sorted(panel_ids): f = evidence_files[candidate_id] @@ -244,7 +256,6 @@ def build_batch_pack( manifest["evidence_count"] = len(evidence_hashes) manifest["evidence_certificate_sha256"] = evidence_hashes - # 3. Safe review references. These are not operational wet-lab protocols. protocol_lines = [ "# Lab Batch Pack — Safe Review References", "", @@ -257,7 +268,6 @@ def build_batch_pack( manifest["included_protocols"].append(str(doc_p)) zf.writestr("protocols/README.md", "\n".join(protocol_lines) + "\n") - # 4. Controls manifest ctrl = controls or [ { "candidate_id": "SEED-001_VAR_064", @@ -270,7 +280,6 @@ def build_batch_pack( zf.writestr("controls_manifest.json", json.dumps(ctrl, indent=2)) manifest["controls"] = ctrl - # 5. Data return template (empty JSON per schema) if schema_p.exists(): template: dict[str, Any] = { "result_id": "REPLACE_WITH_UUID", @@ -290,7 +299,6 @@ def build_batch_pack( zf.writestr("data_return_template.json", json.dumps(template, indent=2)) manifest["data_return_template"] = str(schema_p) - # 6. Chain-of-custody identity hashes custody = { "chain_of_custody_version": "1.1", "scope": ( @@ -313,7 +321,6 @@ def build_batch_pack( zf.writestr("chain_of_custody.json", json.dumps(custody, indent=2)) zf.writestr("MANIFEST.json", json.dumps(manifest, indent=2)) - # 7. Batch README readme = [ "# OpenAMP Foundry — Lab Batch Pack", "", @@ -364,12 +371,17 @@ def build_batch_pack( def main() -> int: import argparse + parser = argparse.ArgumentParser(description="Build lab batch pack zip") parser.add_argument("--panel-csv", default="outputs/wave1_final_panel.csv") parser.add_argument("--evidence-dir", default="outputs/evidence_wave0_5") parser.add_argument("--out", default="outputs/lab_batch_pack.zip") parser.add_argument("--manifest-out", default=None, help="JSON manifest path") - parser.add_argument("--verify-pack", default=None, help="Verify an existing lab batch pack zip") + parser.add_argument( + "--verify-pack", + default=None, + help="Verify an existing lab batch pack zip", + ) args = parser.parse_args() if args.verify_pack: From 73ccf122f63d25dde677e22a09417728f5b515d8 Mon Sep 17 00:00:00 2001 From: Lumi AI Workspace Date: Tue, 7 Jul 2026 00:08:34 +0700 Subject: [PATCH 4/4] Fix lint in lab batch pack tests --- tests/lab/test_lab_batch_pack.py | 146 ++++++++++++++++++++----------- 1 file changed, 97 insertions(+), 49 deletions(-) diff --git a/tests/lab/test_lab_batch_pack.py b/tests/lab/test_lab_batch_pack.py index 496cba3d..f2e8116f 100644 --- a/tests/lab/test_lab_batch_pack.py +++ b/tests/lab/test_lab_batch_pack.py @@ -9,7 +9,10 @@ _REPO_ROOT = Path(__file__).resolve().parents[2] _SCRIPT_PATH = _REPO_ROOT / "scripts" / "lab" / "build_lab_batch_pack.py" -_spec = importlib.util.spec_from_file_location("_scripts_lab_batch_pack", _SCRIPT_PATH) +_spec = importlib.util.spec_from_file_location( + "_scripts_lab_batch_pack", + _SCRIPT_PATH, +) _mod = importlib.util.module_from_spec(_spec) assert _spec.loader is not None _spec.loader.exec_module(_mod) @@ -40,26 +43,43 @@ def _make_panel_csv(tmp: Path, rows: list[tuple[str, str]] | None = None) -> Pat return p -def test_build_pack_success(tmp_path): - panel = _make_panel_csv(tmp_path) - ev_dir = _make_evidence(tmp_path, "C1", "C2") - out = tmp_path / "pack.zip" - result = build_batch_pack( +def _build_pack(panel: Path, ev_dir: Path, out: Path) -> dict: + return build_batch_pack( panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out), ) + + +def _run_pack_cli(*args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, "scripts/lab/build_lab_batch_pack.py", *args], + capture_output=True, + text=True, + env={"PYTHONPATH": "src"}, + ) + + +def test_build_pack_success(tmp_path): + panel = _make_panel_csv(tmp_path) + ev_dir = _make_evidence(tmp_path, "C1", "C2") + out = tmp_path / "pack.zip" + + result = _build_pack(panel, ev_dir, out) + assert "error" not in result assert out.exists() assert result["evidence_count"] == 2 - assert result["strict_identity_checks"]["evidence_must_match_panel_exactly"] is True + assert result["strict_identity_checks"]["evidence_must_match_panel_exactly"] def test_build_pack_contains_evidence(tmp_path): panel = _make_panel_csv(tmp_path) ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" - build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) + + _build_pack(panel, ev_dir, out) + with zipfile.ZipFile(out) as z: names = z.namelist() assert "evidence/C1.json" in names @@ -70,7 +90,9 @@ def test_build_pack_contains_readme(tmp_path): panel = _make_panel_csv(tmp_path) ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" - build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) + + _build_pack(panel, ev_dir, out) + with zipfile.ZipFile(out) as z: assert "README.md" in z.namelist() readme = z.read("README.md").decode("utf-8") @@ -82,7 +104,9 @@ def test_build_pack_contains_controls_manifest(tmp_path): panel = _make_panel_csv(tmp_path) ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" - build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) + + _build_pack(panel, ev_dir, out) + with zipfile.ZipFile(out) as z: assert "controls_manifest.json" in z.namelist() @@ -91,7 +115,9 @@ def test_build_pack_contains_data_return_template(tmp_path): panel = _make_panel_csv(tmp_path) ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" - build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) + + _build_pack(panel, ev_dir, out) + with zipfile.ZipFile(out) as z: assert "data_return_template.json" in z.namelist() @@ -100,7 +126,9 @@ def test_build_pack_contains_chain_of_custody(tmp_path): panel = _make_panel_csv(tmp_path) ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" - manifest = build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) + + manifest = _build_pack(panel, ev_dir, out) + with zipfile.ZipFile(out) as z: custody = json.loads(z.read("chain_of_custody.json")) assert "MANIFEST.json" in z.namelist() @@ -117,8 +145,10 @@ def test_verify_batch_pack_success(tmp_path): panel = _make_panel_csv(tmp_path) ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" - build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) + + _build_pack(panel, ev_dir, out) result = verify_batch_pack(out) + assert result["status"] == "ok" assert result["panel_hash_ok"] is True assert result["candidate_identity_hashes_ok"] is True @@ -130,17 +160,20 @@ def test_verify_batch_pack_detects_tampered_panel(tmp_path): panel = _make_panel_csv(tmp_path) ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" - build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) + + _build_pack(panel, ev_dir, out) tampered = tmp_path / "tampered.zip" - with zipfile.ZipFile(out) as src, zipfile.ZipFile(tampered, "w", zipfile.ZIP_DEFLATED) as dst: - for name in src.namelist(): - data = src.read(name) - if name == "panel.csv": - data = data.replace(b"C1,KKLFKKILKYL", b"C1,KKLFKKILKKK") - dst.writestr(name, data) + with zipfile.ZipFile(out) as src: + with zipfile.ZipFile(tampered, "w", zipfile.ZIP_DEFLATED) as dst: + for name in src.namelist(): + data = src.read(name) + if name == "panel.csv": + data = data.replace(b"C1,KKLFKKILKYL", b"C1,KKLFKKILKKK") + dst.writestr(name, data) result = verify_batch_pack(tampered) + assert result["status"] == "failed" assert result["panel_hash_ok"] is False assert result["candidate_identity_hashes_ok"] is False @@ -151,7 +184,9 @@ def test_build_pack_contains_real_review_reference_docs(tmp_path): panel = _make_panel_csv(tmp_path) ev_dir = _make_evidence(tmp_path, "C1", "C2") out = tmp_path / "pack.zip" - result = build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) + + result = _build_pack(panel, ev_dir, out) + assert "error" not in result with zipfile.ZipFile(out) as z: names = z.namelist() @@ -165,11 +200,13 @@ def test_build_pack_fails_for_missing_required_panel_column(tmp_path): panel = tmp_path / "panel.csv" panel.write_text("id,seq\nC1,KKLFKKILKYL\n") ev_dir = _make_evidence(tmp_path, "C1") + result = build_batch_pack( panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(tmp_path / "pack.zip"), ) + assert "error" in result assert "missing required column" in result["error"] @@ -177,11 +214,13 @@ def test_build_pack_fails_for_missing_required_panel_column(tmp_path): def test_build_pack_fails_for_missing_evidence_certificate(tmp_path): panel = _make_panel_csv(tmp_path) ev_dir = _make_evidence(tmp_path, "C1") + result = build_batch_pack( panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(tmp_path / "pack.zip"), ) + assert "error" in result assert result["missing_evidence"] == ["C2"] @@ -189,11 +228,13 @@ def test_build_pack_fails_for_missing_evidence_certificate(tmp_path): def test_build_pack_fails_for_extra_evidence_certificate(tmp_path): panel = _make_panel_csv(tmp_path, rows=[("C1", "KKLFKKILKYL")]) ev_dir = _make_evidence(tmp_path, "C1", "C2") + result = build_batch_pack( panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(tmp_path / "pack.zip"), ) + assert "error" in result assert result["extra_evidence"] == ["C2"] @@ -201,12 +242,14 @@ def test_build_pack_fails_for_extra_evidence_certificate(tmp_path): def test_build_pack_fails_for_missing_protocol_doc(tmp_path): panel = _make_panel_csv(tmp_path) ev_dir = _make_evidence(tmp_path, "C1", "C2") + result = build_batch_pack( panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(tmp_path / "pack.zip"), protocol_docs=[str(tmp_path / "missing.md")], ) + assert "error" in result assert "Missing protocol/reference document" in result["error"] @@ -233,14 +276,16 @@ def test_cli_runs(tmp_path): ev_dir = _make_evidence(tmp_path, "C1", "C2") panel = _make_panel_csv(tmp_path) out = tmp_path / "pack.zip" - result = subprocess.run( - [sys.executable, "scripts/lab/build_lab_batch_pack.py", - "--panel-csv", str(panel), - "--evidence-dir", str(ev_dir), - "--out", str(out)], - capture_output=True, text=True, - env={"PYTHONPATH": "src"}, + + result = _run_pack_cli( + "--panel-csv", + str(panel), + "--evidence-dir", + str(ev_dir), + "--out", + str(out), ) + assert result.returncode == 0 assert out.exists() @@ -250,15 +295,18 @@ def test_cli_manifest_output(tmp_path): panel = _make_panel_csv(tmp_path) out = tmp_path / "pack.zip" manifest = tmp_path / "manifest.json" - result = subprocess.run( - [sys.executable, "scripts/lab/build_lab_batch_pack.py", - "--panel-csv", str(panel), - "--evidence-dir", str(ev_dir), - "--out", str(out), - "--manifest-out", str(manifest)], - capture_output=True, text=True, - env={"PYTHONPATH": "src"}, + + result = _run_pack_cli( + "--panel-csv", + str(panel), + "--evidence-dir", + str(ev_dir), + "--out", + str(out), + "--manifest-out", + str(manifest), ) + assert result.returncode == 0 assert manifest.exists() data = json.loads(manifest.read_text()) @@ -272,14 +320,16 @@ def test_cli_returns_2_for_missing_evidence(tmp_path): ev_dir = _make_evidence(tmp_path, "C1") panel = _make_panel_csv(tmp_path) out = tmp_path / "pack.zip" - result = subprocess.run( - [sys.executable, "scripts/lab/build_lab_batch_pack.py", - "--panel-csv", str(panel), - "--evidence-dir", str(ev_dir), - "--out", str(out)], - capture_output=True, text=True, - env={"PYTHONPATH": "src"}, + + result = _run_pack_cli( + "--panel-csv", + str(panel), + "--evidence-dir", + str(ev_dir), + "--out", + str(out), ) + assert result.returncode == 2 assert "Missing evidence certificate" in result.stderr @@ -288,12 +338,10 @@ def test_cli_verify_pack(tmp_path): ev_dir = _make_evidence(tmp_path, "C1", "C2") panel = _make_panel_csv(tmp_path) out = tmp_path / "pack.zip" - build_batch_pack(panel_csv=str(panel), evidence_dir=str(ev_dir), out_zip=str(out)) - result = subprocess.run( - [sys.executable, "scripts/lab/build_lab_batch_pack.py", "--verify-pack", str(out)], - capture_output=True, text=True, - env={"PYTHONPATH": "src"}, - ) + + _build_pack(panel, ev_dir, out) + result = _run_pack_cli("--verify-pack", str(out)) + assert result.returncode == 0 payload = json.loads(result.stdout) assert payload["status"] == "ok"