feat: Phase V V2 evidence bundle manifest (EBM-) -- portable artifact index for external handoff

src/openamp_foundry/evidence/evidence_bundle_manifest.py

@@ -0,0 +1,141 @@
+"""EBM- evidence bundle manifest schema.
+
+Portable listing of all evidence artifact IDs and schema types for a batch.
+Enables reproducibility checks and external handoff; machine-readable index
+a scientist can use to request any artifact.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+VALID_MANIFEST_SCHEMA_TYPES: frozenset[str] = frozenset({
+    "BSP", "PSC", "BOS", "CPS", "CBA", "CRG",
+    "RSR", "PQT", "BTI", "CBA2", "BEG", "SAT", "PCC",
+    "FET", "BRC", "CSV", "PTR", "ECI", "PRG",
+})
+
+VALID_MANIFEST_STATUSES: frozenset[str] = frozenset({
+    "complete", "partial", "empty",
+})
+
+
+@dataclass
+class ManifestEntry:
+    schema_type: str
+    artifact_id: str
+    description: str
+
+
+@dataclass
+class EvidenceBundleManifest:
+    ebm_id: str
+    batch_id: str
+    pipeline_version: str
+    entries: list[ManifestEntry]
+    n_entries: int
+    schema_types_included: list[str]
+    manifest_status: str
+    dry_lab_only: bool
+    limitations: list[str]
+    created_at: str
+
+
+def _compute_status(n_entries: int) -> str:
+    if n_entries == 0:
+        return "empty"
+    n_required = len(VALID_MANIFEST_SCHEMA_TYPES)
+    if n_entries >= n_required:
+        return "complete"
+    return "partial"
+
+
+def validate_evidence_bundle_manifest(ebm: EvidenceBundleManifest) -> None:
+    if not ebm.ebm_id.startswith("EBM-"):
+        raise ValueError(f"ebm_id must start with 'EBM-': {ebm.ebm_id!r}")
+    if not ebm.batch_id:
+        raise ValueError("batch_id must be non-empty")
+    if not ebm.pipeline_version:
+        raise ValueError("pipeline_version must be non-empty")
+    for entry in ebm.entries:
+        if entry.schema_type not in VALID_MANIFEST_SCHEMA_TYPES:
+            raise ValueError(
+                f"schema_type {entry.schema_type!r} not in VALID_MANIFEST_SCHEMA_TYPES"
+            )
+        if not entry.artifact_id:
+            raise ValueError(
+                f"artifact_id must be non-empty for schema_type {entry.schema_type!r}"
+            )
+    if ebm.manifest_status not in VALID_MANIFEST_STATUSES:
+        raise ValueError(
+            f"manifest_status {ebm.manifest_status!r} not in VALID_MANIFEST_STATUSES"
+        )
+    if not ebm.dry_lab_only:
+        raise ValueError("dry_lab_only must be True")
+    if not ebm.limitations:
+        raise ValueError("limitations must be non-empty")
+    if not ebm.created_at:
+        raise ValueError("created_at must be non-empty")
+    if ebm.n_entries != len(ebm.entries):
+        raise ValueError("n_entries must equal len(entries)")
+    expected_types = sorted({e.schema_type for e in ebm.entries})
+    if ebm.schema_types_included != expected_types:
+        raise ValueError("schema_types_included mismatch")
+
+
+def build_evidence_bundle_manifest(
+    *,
+    ebm_id: str,
+    batch_id: str,
+    pipeline_version: str,
+    entry_dicts: list[dict],
+    limitations: list[str],
+    created_at: str,
+) -> EvidenceBundleManifest:
+    """Build an evidence bundle manifest.
+
+    entry_dicts: list of dicts with keys: schema_type, artifact_id, description.
+    """
+    entries = [
+        ManifestEntry(
+            schema_type=d["schema_type"],
+            artifact_id=d["artifact_id"],
+            description=d.get("description", ""),
+        )
+        for d in entry_dicts
+    ]
+    schema_types = sorted({e.schema_type for e in entries})
+    status = _compute_status(len(schema_types))
+
+    ebm = EvidenceBundleManifest(
+        ebm_id=ebm_id,
+        batch_id=batch_id,
+        pipeline_version=pipeline_version,
+        entries=entries,
+        n_entries=len(entries),
+        schema_types_included=schema_types,
+        manifest_status=status,
+        dry_lab_only=True,
+        limitations=limitations,
+        created_at=created_at,
+    )
+    validate_evidence_bundle_manifest(ebm)
+    return ebm
+
+
+def format_evidence_bundle_manifest(ebm: EvidenceBundleManifest) -> str:
+    lines = [
+        f"Evidence Bundle Manifest — {ebm.ebm_id}",
+        f"Batch: {ebm.batch_id}  |  Pipeline: {ebm.pipeline_version}",
+        f"Status: {ebm.manifest_status}  |  Entries: {ebm.n_entries}",
+        f"Schema types: {', '.join(ebm.schema_types_included)}",
+    ]
+    if ebm.entries:
+        lines.append("Artifacts:")
+        for entry in ebm.entries:
+            desc = f" — {entry.description}" if entry.description else ""
+            lines.append(f"  {entry.schema_type}: {entry.artifact_id}{desc}")
+    lines.append(f"Created: {ebm.created_at}")
+    lines.append(f"Limitations: {'; '.join(ebm.limitations)}")
+    lines.append(f"dry_lab_only: {ebm.dry_lab_only}")
+    return "\n".join(lines)

tests/evidence/test_evidence_bundle_manifest.py

@@ -0,0 +1,255 @@
+"""Tests for EBM- evidence bundle manifest schema."""
+
+import pytest
+from openamp_foundry.evidence.evidence_bundle_manifest import (
+    EvidenceBundleManifest,
+    ManifestEntry,
+    VALID_MANIFEST_SCHEMA_TYPES,
+    VALID_MANIFEST_STATUSES,
+    build_evidence_bundle_manifest,
+    format_evidence_bundle_manifest,
+    validate_evidence_bundle_manifest,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_TWO_ENTRIES = [
+    {"schema_type": "BSP", "artifact_id": "A1", "description": "batch selection"},
+    {"schema_type": "PCC", "artifact_id": "A2", "description": "completeness cert"},
+]
+
+
+def _build(**kwargs):
+    defaults = dict(
+        ebm_id="EBM-001",
+        batch_id="BATCH-01",
+        pipeline_version="v1.0",
+        entry_dicts=_TWO_ENTRIES,
+        limitations=["dry-lab only"],
+        created_at="2026-07-10",
+    )
+    defaults.update(kwargs)
+    return build_evidence_bundle_manifest(**defaults)
+
+
+# ---------------------------------------------------------------------------
+# 1. Constants
+# ---------------------------------------------------------------------------
+
+
+def test_valid_manifest_schema_types_is_frozenset():
+    assert isinstance(VALID_MANIFEST_SCHEMA_TYPES, frozenset)
+
+
+def test_valid_manifest_schema_types_contains_bsp():
+    assert "BSP" in VALID_MANIFEST_SCHEMA_TYPES
+
+
+def test_valid_manifest_schema_types_contains_pcc():
+    assert "PCC" in VALID_MANIFEST_SCHEMA_TYPES
+
+
+def test_valid_manifest_schema_types_contains_prg():
+    assert "PRG" in VALID_MANIFEST_SCHEMA_TYPES
+
+
+def test_valid_manifest_schema_types_contains_eci():
+    assert "ECI" in VALID_MANIFEST_SCHEMA_TYPES
+
+
+def test_valid_manifest_schema_types_count():
+    assert len(VALID_MANIFEST_SCHEMA_TYPES) == 19
+
+
+def test_valid_manifest_statuses_is_frozenset():
+    assert isinstance(VALID_MANIFEST_STATUSES, frozenset)
+
+
+def test_valid_manifest_statuses_contains_complete():
+    assert "complete" in VALID_MANIFEST_STATUSES
+
+
+def test_valid_manifest_statuses_contains_partial():
+    assert "partial" in VALID_MANIFEST_STATUSES
+
+
+def test_valid_manifest_statuses_contains_empty():
+    assert "empty" in VALID_MANIFEST_STATUSES
+
+
+# ---------------------------------------------------------------------------
+# 2. build – happy paths
+# ---------------------------------------------------------------------------
+
+
+def test_build_returns_evidence_bundle_manifest():
+    assert isinstance(_build(), EvidenceBundleManifest)
+
+
+def test_build_ebm_id_stored():
+    assert _build().ebm_id == "EBM-001"
+
+
+def test_build_batch_id_stored():
+    assert _build().batch_id == "BATCH-01"
+
+
+def test_build_pipeline_version_stored():
+    assert _build().pipeline_version == "v1.0"
+
+
+def test_build_dry_lab_only_true():
+    assert _build().dry_lab_only is True
+
+
+def test_build_n_entries_matches_input():
+    assert _build().n_entries == 2
+
+
+def test_build_partial_status_when_subset():
+    assert _build().manifest_status == "partial"
+
+
+def test_build_empty_status_when_no_entries():
+    r = _build(entry_dicts=[])
+    assert r.manifest_status == "empty"
+
+
+def test_build_schema_types_included_sorted():
+    r = _build()
+    assert r.schema_types_included == sorted(r.schema_types_included)
+
+
+def test_build_schema_types_deduplicates():
+    entries = [
+        {"schema_type": "BSP", "artifact_id": "A1", "description": "d1"},
+        {"schema_type": "BSP", "artifact_id": "A2", "description": "d2"},
+    ]
+    r = _build(entry_dicts=entries)
+    assert r.schema_types_included == ["BSP"]
+    assert r.n_entries == 2
+
+
+def test_build_entries_are_manifest_entry():
+    for e in _build().entries:
+        assert isinstance(e, ManifestEntry)
+
+
+def test_build_entry_artifact_id_stored():
+    r = _build()
+    assert r.entries[0].artifact_id == "A1"
+
+
+def test_build_entry_description_stored():
+    r = _build()
+    assert r.entries[0].description == "batch selection"
+
+
+def test_build_empty_description_defaults_to_empty_string():
+    entries = [{"schema_type": "BSP", "artifact_id": "A1"}]
+    r = _build(entry_dicts=entries)
+    assert r.entries[0].description == ""
+
+
+def test_build_limitations_stored():
+    assert _build().limitations == ["dry-lab only"]
+
+
+def test_build_created_at_stored():
+    assert _build().created_at == "2026-07-10"
+
+
+def test_build_complete_status_when_all_types_present():
+    all_entries = [
+        {"schema_type": t, "artifact_id": f"ART-{t}", "description": ""}
+        for t in VALID_MANIFEST_SCHEMA_TYPES
+    ]
+    r = _build(entry_dicts=all_entries)
+    assert r.manifest_status == "complete"
+
+
+# ---------------------------------------------------------------------------
+# 3. validate – rejection cases
+# ---------------------------------------------------------------------------
+
+
+def test_validate_rejects_bad_ebm_id_prefix():
+    with pytest.raises(ValueError, match="EBM-"):
+        _build(ebm_id="BAD-001")
+
+
+def test_validate_rejects_empty_batch_id():
+    with pytest.raises(ValueError):
+        _build(batch_id="")
+
+
+def test_validate_rejects_empty_pipeline_version():
+    with pytest.raises(ValueError):
+        _build(pipeline_version="")
+
+
+def test_validate_rejects_unknown_schema_type():
+    entries = [{"schema_type": "UNKNOWN", "artifact_id": "A1", "description": "x"}]
+    with pytest.raises(ValueError, match="VALID_MANIFEST_SCHEMA_TYPES"):
+        _build(entry_dicts=entries)
+
+
+def test_validate_rejects_empty_artifact_id():
+    entries = [{"schema_type": "BSP", "artifact_id": "", "description": "x"}]
+    with pytest.raises(ValueError, match="artifact_id"):
+        _build(entry_dicts=entries)
+
+
+def test_validate_rejects_empty_limitations():
+    with pytest.raises(ValueError, match="limitations"):
+        _build(limitations=[])
+
+
+def test_validate_rejects_empty_created_at():
+    with pytest.raises(ValueError):
+        _build(created_at="")
+
+
+# ---------------------------------------------------------------------------
+# 4. format
+# ---------------------------------------------------------------------------
+
+
+def test_format_contains_ebm_id():
+    assert "EBM-001" in format_evidence_bundle_manifest(_build())
+
+
+def test_format_contains_batch_id():
+    assert "BATCH-01" in format_evidence_bundle_manifest(_build())
+
+
+def test_format_contains_status():
+    assert "partial" in format_evidence_bundle_manifest(_build())
+
+
+def test_format_contains_schema_types():
+    formatted = format_evidence_bundle_manifest(_build())
+    assert "BSP" in formatted
+    assert "PCC" in formatted
+
+
+def test_format_contains_artifact_ids():
+    assert "A1" in format_evidence_bundle_manifest(_build())
+
+
+def test_format_contains_description():
+    assert "batch selection" in format_evidence_bundle_manifest(_build())
+
+
+def test_format_contains_limitations():
+    assert "dry-lab only" in format_evidence_bundle_manifest(_build())
+
+
+def test_format_contains_dry_lab_only():
+    assert "dry_lab_only: True" in format_evidence_bundle_manifest(_build())
+
+
+def test_format_is_string():
+    assert isinstance(format_evidence_bundle_manifest(_build()), str)