cjm-provenance-bundle 0.0.4

Generic provenance bundle envelope: typed hash-linked stage chains, payload staging and relocation, reference-vs-copy dispositions, integrity/chain validation, and atomic packaging for sharing workflow outputs with verifiable provenance.

cjm-provenance-bundle

Design notes

Named CD2 exception (day-one extraction). This library was extracted ahead of multi-consumer evidence, deliberately: chain validation and payload integrity are security-sensitive primitives that belong in one well-reviewed implementation, and the bundle envelope is the cross-workflow data-interchange contract — its evidence is the multi-workflow pipeline’s existence, not the count of consumers inside it. (Precedent: the V12 design-system/app-core layering exception.)

Wrap, with the chain-link grammar absorbed (ratified 2026-06-11). The envelope owns the typed per-stage chain record — format, version, payload hash, upstream link by hash, never by path — plus integrity, staging/relocation, and tolerant-unknown round-trip. The workflow cores’ run manifests stay authoritative domain payloads, staged verbatim (hash-stable, never rewritten); the envelope’s payload entries are the relocation map. Per-source-type format knowledge lives in adapter libraries (e.g. cjm-provenance-bundle-transcript), mirroring the neutral-grammar/domain-schema split used by cjm-context-graph-layer / cjm-transcript-graph-schema.

Reference-vs-copy is a per-payload-class disposition keyed on re-derivability — and for source media, a rights boundary. Referenced payloads carry locator + content hash without the bytes; recipients bind their own licensed copy, which verifies (or refuses) by hash. Import honesty is three-way: bundle-internal missing/corrupt = loud error; declared-external unresolvable = pending verification; external resolvable but mismatched = loud error.

Install

pip install cjm_provenance_bundle

Project Structure

nbs/
├── adapter.ipynb   # The source-type adapter seam: the envelope owns trust (chain, hashes, staging, packaging); a per-source-type adapter owns format interpretation (which fields are stageable artifacts, where the upstream pointer lives). Deliberately tiny -- `extract_stage` + the concrete `walk_chain` -- so new source types add format knowledge without touching trust code (the grammar-lib/schema-lib split, one level up). `export_bundle` is the generic export orchestrator.
├── cli.ipynb       # The generic, adapter-independent console driver: `inspect` / `verify` / `extract` / `bind` -- the trust operations every bundle supports regardless of source type. `export`/`import` live with the source-type adapter libraries (they need format knowledge + a capability runtime); see `cjm-provenance-bundle-transcript`.
├── integrity.ipynb # Chain + payload verification: the trust core of the envelope (CD2 day-one extraction exception -- security-sensitive validation written once, reviewed once). Implements the three-way import honesty ratified 2026-06-11: bundle-internal missing/corrupt = LOUD error; declared-external unresolvable = PENDING (verify later via bind); external resolvable but mismatched = LOUD error.
├── manifest.ipynb  # The bundle envelope schema (CR-20): typed, hash-linked stage chains wrapping the workflow cores' run manifests as verbatim payloads. The envelope owns the chain-link grammar (format/version/payload-hash/upstream-link-BY-HASH); domain payloads stay authoritative in their producing cores (wrap-with-absorbed-chain-grammar, ratified 2026-06-11). Strict-known / tolerant-unknown at every level (P5/P6 law).
├── reader.ipynb    # Import-side bundle access: open (zip or extracted dir), envelope-validate (a half-bundle or tampered envelope refuses at open -- bundle.json + sidecar are the gate), verify payloads, resolve through the relocation map, and `bind` local copies of referenced externals (hash-verified; the rights-boundary flow). Zip extraction guards against path traversal (the SG-1 tar lesson applied to zip).
└── staging.ipynb   # Export-side payload staging + atomic packaging: stage files into a working directory (copied verbatim -- hash-stable), then finalize as `bundle.json` + sidecar + ZIP via temp-name + fsync + atomic rename. An interrupted export leaves only a `.partial-*` temp file; no half-bundle can ever be taken for a bundle (ratified stress item 7 by construction).

Total: 6 notebooks

Module Dependencies

graph LR
    adapter["adapter<br/>adapter"]
    cli["cli<br/>cli"]
    integrity["integrity<br/>integrity"]
    manifest["manifest<br/>manifest"]
    reader["reader<br/>reader"]
    staging["staging<br/>staging"]

    adapter --> manifest
    adapter --> staging
    adapter --> reader
    cli --> manifest
    cli --> reader
    cli --> adapter
    integrity --> manifest
    reader --> integrity
    reader --> manifest
    reader --> staging
    staging --> manifest
    staging --> integrity

12 cross-module dependencies detected

CLI Reference

cjm-provenance-bundle Command

usage: cjm-provenance-bundle [-h] [-v] {inspect,verify,extract,bind} ...

Inspect / verify / extract / bind provenance bundles.

positional arguments:
  {inspect,verify,extract,bind}
    inspect             Print the bundle summary (envelope-validates only)
    verify              Verify chain + every payload (three-way external
                        honesty)
    extract             Extract a bundle zip to a directory + verify
    bind                Hash-verify a local file against a referenced payload

options:
  -h, --help            show this help message and exit
  -v, --verbose         DEBUG-level logging

For detailed help on any command, use cjm-provenance-bundle <command> --help.

Module Overview

Detailed documentation for each module in the project:

adapter (adapter.ipynb)

The source-type adapter seam: the envelope owns trust (chain, hashes, staging, packaging); a per-source-type adapter owns format interpretation (which fields are stageable artifacts, where the upstream pointer lives). Deliberately tiny – extract_stage + the concrete walk_chain – so new source types add format knowledge without touching trust code (the grammar-lib/schema-lib split, one level up). export_bundle is the generic export orchestrator.

Import

from cjm_provenance_bundle.adapter import (
    ArtifactSpec,
    StageExtraction,
    SourceTypeAdapter,
    register_adapter,
    get_adapter,
    adapter_for_format,
    export_bundle
)

Functions

def register_adapter(
    adapter: SourceTypeAdapter,  # Adapter instance
) -> SourceTypeAdapter:  # The same instance (decorator-friendly)
    "Register an adapter under its source_type."

def get_adapter(
    source_type: str,  # Adapter discriminator
) -> SourceTypeAdapter:  # The registered adapter
    "Look up a registered adapter; loud on absence (the adapter library is not installed)."

def adapter_for_format(
    fmt: str,  # A run-manifest format tag
) -> SourceTypeAdapter:  # The adapter that reads it
    "Find the registered adapter that knows a run-manifest format."

def _package_version() -> str:  # This library's installed version (best effort)
    """Resolve the installed cjm-provenance-bundle version for exporter identity."""
    try
    "Resolve the installed cjm-provenance-bundle version for exporter identity."

def export_bundle(
    terminal_manifests: Sequence[Union[str, Path]],   # One or more terminal run manifests (chains may share roots)
    adapter: SourceTypeAdapter,                       # The source-type adapter
    out_path: Union[str, Path],                       # Destination .zip
    options: Optional[Dict[str, Any]] = None,         # Export options (adapter-interpreted)
    attachments: Sequence[Tuple[Union[str, Path], str]] = (),  # Bundle-level payloads: (path, payload_class)
    exporter_extra: Optional[Dict[str, Any]] = None,  # Extra exporter-identity fields (e.g. adapter version)
    work_dir: Optional[Union[str, Path]] = None,      # Staging dir override
) -> Path:  # The written bundle path
    """
    Walk each chain, stage manifests verbatim + classified artifacts, hash-link the
    stages, attach bundle-level payloads, and finalize atomically. Stages reached from
    multiple terminals are deduplicated by resolved path.
    """

Classes

@dataclass
class ArtifactSpec:
    "An artifact a stage contributes, as classified by the source-type adapter."
    
    path: str  # Local path at export time
    payload_class: str  # Semantic class (e.g. "source-media")
    disposition: str  # DISPOSITION_COPIED | DISPOSITION_REFERENCED
    known_hash: Optional[str]  # Hash recorded in the run manifest (export-time honesty + absent-file referencing)

@dataclass
class StageExtraction:
    "Everything the envelope needs from one wrapped run manifest."
    
    stage_id: str  # The run id
    format: str  # The manifest's format tag
    version: str  # The manifest's schema version
    created_at: float  # The run's creation timestamp
    upstream_path: Optional[str]  # Local path of the consumed upstream manifest (None = chain root)
    artifacts: List[ArtifactSpec] = field(...)  # Classified artifacts
    display: Dict[str, Any] = field(...)  # Human-readable stage summary

class SourceTypeAdapter(ABC):
    """
    Format knowledge for one source type (CR-20 Option-C bundle adapter).
    
    Subclasses declare `source_type` + `known_formats` and implement `extract_stage`;
    the chain walk and all trust operations are generic.
    """
    
    def extract_stage(
            self,
            manifest: Dict[str, Any],        # Parsed run-manifest dict
            manifest_path: str,              # Its local path (for resolving relative pointers)
            options: Dict[str, Any],         # Export options (e.g. {"media_disposition": "referenced"})
        ) -> StageExtraction:  # The envelope-facing extraction
        "Classify one run manifest (upstream pointer + artifact specs + display)."
    
    def load_manifest(
            self,
            path: Union[str, Path],  # Run-manifest path
        ) -> Dict[str, Any]:  # Parsed dict (format-gated)
        "Load + gate a run manifest against this adapter's known formats."
    
    def walk_chain(
        "Walk upstream pointers from terminal to root (cycle-guarded), return root-first."

Variables

_ADAPTERS: Dict[str, SourceTypeAdapter]

cli (cli.ipynb)

The generic, adapter-independent console driver: inspect / verify / extract / bind – the trust operations every bundle supports regardless of source type. export/import live with the source-type adapter libraries (they need format knowledge + a capability runtime); see cjm-provenance-bundle-transcript.

Import

from cjm_provenance_bundle.cli import (
    logger,
    print_summary,
    print_report,
    build_parser,
    main
)

Functions

def print_summary(
    reader: BundleReader,  # Open bundle
) -> None
    "Print a human-readable bundle summary."

def print_report(
    report,  # VerificationReport
) -> None
    "Print a verification report (errors loud, pendings informational)."

def build_parser() -> argparse.ArgumentParser:  # Configured CLI parser
    "Build the generic bundle CLI (trust operations only)."

def main(
    argv: Optional[List[str]] = None,  # CLI args (default: sys.argv[1:])
) -> int:  # Process exit code
    "Generic bundle CLI entry point."

integrity (integrity.ipynb)

Chain + payload verification: the trust core of the envelope (CD2 day-one extraction exception – security-sensitive validation written once, reviewed once). Implements the three-way import honesty ratified 2026-06-11: bundle-internal missing/corrupt = LOUD error; declared-external unresolvable = PENDING (verify later via bind); external resolvable but mismatched = LOUD error.

Import

from cjm_provenance_bundle.integrity import (
    SIDECAR_SUFFIX,
    PendingExternal,
    VerificationReport,
    verify_chain,
    resolve_external_candidate,
    verify_bundle,
    write_sidecar,
    check_sidecar
)

Functions

def verify_chain(
    bundle: BundleManifest,  # Parsed envelope manifest
) -> List[str]:  # Structural chain errors (empty = chain sound)
    """
    Validate the hash-linked chain structurally (no file I/O).
    
    Every upstream link must resolve to a stage manifest hash present in the bundle;
    stage ids and manifest hashes must be unique; at least one root must exist; every
    stage manifest must be a copied payload (the chain must be self-contained).
    """

def _hash_with_algo_of(
    path: Union[str, Path],  # File to hash
    expected: str,           # Expected "algo:hexdigest" (selects the algorithm)
) -> str:  # Actual hash in the same algo
    "Hash a file with the algorithm named by the expected hash string."

def resolve_external_candidate(
    entry: PayloadEntry,                        # A referenced payload entry
    bindings: Optional[Dict[str, str]] = None,  # content_hash -> locally bound path
) -> Optional[str]:  # Candidate local path (unverified), or None
    """
    Find a local candidate for a referenced payload: explicit binding first, then the
    original FileRef location if it still exists. Returns a path WITHOUT hashing it --
    verification hashes exactly once at the call site.
    """

def verify_bundle(
    root: Union[str, Path],                     # Extracted bundle root directory
    bundle: BundleManifest,                     # Parsed envelope manifest
    check_external: bool = True,                # Also attempt referenced payloads
    bindings: Optional[Dict[str, str]] = None,  # content_hash -> locally bound path
) -> VerificationReport:  # Full report (errors + pendings + counts)
    """
    Verify every payload + the chain, with three-way external honesty.
    
    copied + missing/mismatched = ERROR (the bundle lied about itself);
    referenced + unresolvable = PENDING (bind later; hash still pins identity);
    referenced + resolvable-but-mismatched = ERROR (wrong or tampered file).
    """

def write_sidecar(
    bundle_json_path: Union[str, Path],  # Written bundle.json path
) -> Path:  # The sidecar path
    "Write the envelope's own integrity sidecar (manifest edits become loud)."

def check_sidecar(
    bundle_json_path: Union[str, Path],  # bundle.json path
) -> None
    "Verify bundle.json against its sidecar; raises loudly on tamper or absence."

Classes

@dataclass
class PendingExternal:
    """
    A referenced payload that could not be resolved locally -- NOT an error.
    The recipient can `bind` a local copy later; the recorded content hash then verifies it.
    """
    
    owner: str  # Human-readable owner ("stage <id>" or "attachment")
    entry: PayloadEntry  # The referenced payload entry
    reason: str  # Why it is pending (for display)

@dataclass
class VerificationReport:
    "Outcome of a full bundle verification."
    
    errors: List[str] = field(...)  # LOUD failures (tamper, missing internals, broken chain)
    pending: List[PendingExternal] = field(...)  # Unresolvable external refs (verify-later)
    payloads_verified: int = 0  # Payloads whose bytes hash-verified
    stages_checked: int = 0  # Stages examined
    
    def ok(self) -> bool:  # True when no loud failures (pendings allowed)
        "Verification passes when there are zero errors; pendings are acceptable."

Variables

SIDECAR_SUFFIX = '.sha256'  # Envelope sidecar: bundle.json.sha256 holds the manifest's own hash

manifest (manifest.ipynb)

The bundle envelope schema (CR-20): typed, hash-linked stage chains wrapping the workflow cores’ run manifests as verbatim payloads. The envelope owns the chain-link grammar (format/version/payload-hash/upstream-link-BY-HASH); domain payloads stay authoritative in their producing cores (wrap-with-absorbed-chain-grammar, ratified 2026-06-11). Strict-known / tolerant-unknown at every level (P5/P6 law).

Import

from cjm_provenance_bundle.manifest import (
    BUNDLE_FORMAT,
    BUNDLE_VERSION,
    DISPOSITION_COPIED,
    DISPOSITION_REFERENCED,
    DISPOSITIONS,
    PAYLOAD_CLASS_STAGE_MANIFEST,
    BundleError,
    BundleFormatError,
    BundleIntegrityError,
    PayloadEntry,
    StageRecord,
    BundleManifest,
    stage_by_manifest_hash,
    root_stages,
    terminal_stages,
    new_bundle_id
)

Functions

def _split_known(
    d: Dict[str, Any],        # Incoming dict
    known: frozenset,         # Known top-level keys
) -> Dict[str, Any]:  # The unknown-key remainder (tolerant-unknown round-trip)
    "Collect unknown keys so they survive a round-trip through this library version."

def _merge_extras(
    known: Dict[str, Any],    # Serialized known fields
    extras: Dict[str, Any],   # Unknown-key remainder captured at parse time
) -> Dict[str, Any]:  # Merged dict; known fields always win
    "Merge round-tripped unknown keys back in without letting them clobber known fields."

def stage_by_manifest_hash(
    bundle: BundleManifest,  # Parsed envelope manifest
) -> Dict[str, StageRecord]:  # manifest content_hash -> stage
    "Index stages by their wrapped manifest's content hash (the chain-link key)."

def root_stages(
    bundle: BundleManifest,  # Parsed envelope manifest
) -> List[StageRecord]:  # Stages with no upstream (chain roots)
    "The chain roots (a bundle may carry a forest -- e.g. one chain per episode)."

def terminal_stages(
    bundle: BundleManifest,  # Parsed envelope manifest
) -> List[StageRecord]:  # Stages no other stage links upstream to
    "The chain terminals (the manifests the export started from)."

def new_bundle_id() -> str:  # e.g. "bundle_20260611_153000_1a2b3c4d"
    "Generate a unique, sortable bundle id."

Classes

class BundleError(RuntimeError):
    "Base class for provenance-bundle errors."

class BundleFormatError(BundleError):
    """
    The bundle (or a wrapped payload) is structurally unusable: missing/unparseable
    manifest, unknown format tag, unsupported major version. A half-written bundle is
    refused with this error -- never partially accepted.
    """

class BundleIntegrityError(BundleError):
    """
    A trust claim failed LOUDLY: payload hash mismatch (tamper/corruption), broken
    chain link, envelope sidecar mismatch, or a bind against the wrong content.
    """

@dataclass
class PayloadEntry:
    """
    One artifact the bundle knows about -- copied into the bundle, or referenced by
    locator + content hash (identity-vs-location split: the hash verifies regardless of
    whether the locator resolves).
    """
    
    original: Dict[str, Any]  # Locator wire dict for where the artifact lived at export (relocation-map key)
    content_hash: str  # "algo:hexdigest" over the artifact bytes -- present for BOTH dispositions
    disposition: str  # DISPOSITION_COPIED | DISPOSITION_REFERENCED
    payload_class: str  # Semantic class (e.g. "stage-manifest", "source-media", "graph-export")
    bundle_path: Optional[str]  # Bundle-relative path (copied only; None when referenced)
    size_bytes: Optional[int]  # Informational size (None if unknown, e.g. referenced + absent locally)
    extras: Dict[str, Any] = field(...)  # Unknown-key round-trip
    
    def to_dict(self) -> Dict[str, Any]:  # Plain-dict form
            """Serialize; round-tripped unknown keys ride along (known fields win)."""
            known = {
                "original": self.original,
        "Serialize; round-tripped unknown keys ride along (known fields win)."
    
    def from_dict(cls, d: Dict[str, Any]) -> "PayloadEntry":  # Parsed entry
        "Parse; unknown keys are captured losslessly into `extras`."

@dataclass
class StageRecord:
    """
    One hash-linked chain link: a workflow stage whose run manifest is wrapped
    VERBATIM as a copied payload. `upstream` carries the upstream stage manifest's
    content hash -- the chain is a property of the bytes, never of filesystem paths.
    """
    
    stage_id: str  # Producing run id (unique within the bundle)
    format: str  # The wrapped payload's own format tag (e.g. "cjm-transcript-decomp-core/run-manifest")
    version: str  # The wrapped payload's own schema version
    created_at: float  # The wrapped run's creation timestamp
    manifest: PayloadEntry  # The stage's run manifest, staged verbatim (disposition always "copied")
    upstream: Optional[str]  # content_hash of the upstream stage's manifest; None = chain root
    artifacts: List[PayloadEntry] = field(...)  # Artifacts this stage contributes
    extras: Dict[str, Any] = field(...)  # Unknown-key round-trip
    
    def to_dict(self) -> Dict[str, Any]:  # Plain-dict form
            """Serialize with nested payload entries."""
            known = {
                "stage_id": self.stage_id,
        "Serialize with nested payload entries."
    
    def from_dict(cls, d: Dict[str, Any]) -> "StageRecord":  # Parsed record
        "Parse with nested payload entries; unknown keys captured into `extras`."

@dataclass
class BundleManifest:
    """
    The envelope manifest: chain stages (root-first) + bundle-level attachments
    (payloads owned by no single stage, e.g. a typed graph export spanning the chain).
    """
    
    bundle_id: str  # Unique bundle identifier
    exported_at: float  # Export timestamp
    source_type: str  # Source-type adapter discriminator (e.g. "transcript")
    exporter: Dict[str, Any] = field(...)  # Exporting library identity {"library", "version", ...}
    stages: List[StageRecord] = field(...)  # Chain stages, root-first
    attachments: List[PayloadEntry] = field(...)  # Bundle-level payloads
    extras: Dict[str, Any] = field(...)  # Unknown-key round-trip
    FORMAT: str = field(...)  # Envelope format tag
    VERSION: str = field(...)  # Envelope schema version
    
    def to_dict(self) -> Dict[str, Any]:  # Plain-dict form
            """Serialize the full envelope manifest."""
            known = {
                "format": self.FORMAT,
        "Serialize the full envelope manifest."
    
    def to_json(self) -> str:  # Pretty-printed JSON
            """Render the envelope manifest as JSON."""
            return json.dumps(self.to_dict(), indent=2)
    
        @classmethod
        def from_dict(cls, d: Dict[str, Any]) -> "BundleManifest":  # Parsed manifest
        "Render the envelope manifest as JSON."
    
    def from_dict(cls, d: Dict[str, Any]) -> "BundleManifest":  # Parsed manifest
            """Parse + gate the envelope format/version.
    
            Major version mismatch refuses loudly; a NEWER minor version is accepted
            (tolerant-unknown: its additive keys round-trip via `extras`)."""
            fmt = d.get("format", "")
            if fmt != BUNDLE_FORMAT
        "Parse + gate the envelope format/version.

Major version mismatch refuses loudly; a NEWER minor version is accepted
(tolerant-unknown: its additive keys round-trip via `extras`)."
    
    def from_json(cls, text: str) -> "BundleManifest":  # Parsed manifest
            """Parse the envelope manifest from JSON text."""
            try
        "Parse the envelope manifest from JSON text."

Variables

BUNDLE_FORMAT = 'cjm-provenance-bundle/manifest'  # Envelope manifest format tag
BUNDLE_VERSION = '0.1.0'  # Envelope manifest schema version
DISPOSITION_COPIED = 'copied'  # Payload bytes live inside the bundle (relative bundle_path)
DISPOSITION_REFERENCED = 'referenced'  # Locator + content hash only; bytes stay external (rights/size boundary)
DISPOSITIONS
PAYLOAD_CLASS_STAGE_MANIFEST = 'stage-manifest'  # The wrapped run manifest itself (always copied)

reader (reader.ipynb)

Import-side bundle access: open (zip or extracted dir), envelope-validate (a half-bundle or tampered envelope refuses at open – bundle.json + sidecar are the gate), verify payloads, resolve through the relocation map, and bind local copies of referenced externals (hash-verified; the rights-boundary flow). Zip extraction guards against path traversal (the SG-1 tar lesson applied to zip).

Import

from cjm_provenance_bundle.reader import (
    BundleReader,
    open_bundle
)

Functions

def _safe_extract_zip(
    zf: zipfile.ZipFile,       # Open zip
    dest: Union[str, Path],    # Extraction destination
) -> None
    "Extract with a path-traversal guard (zip-slip): every member must resolve under dest."

def open_bundle(
    path: Union[str, Path],                         # Bundle .zip OR an extracted bundle dir
    extract_to: Optional[Union[str, Path]] = None,  # Extraction dest when path is a zip
) -> BundleReader:  # Open reader
    "Open a bundle from either form."

Classes

class BundleReader:
    def __init__(
        self,
        root: Union[str, Path],  # Extracted bundle root (contains bundle.json)
    )
    """
    Read-side handle on an extracted bundle directory.
    
    Opening validates the ENVELOPE only (bundle.json present + parseable + sidecar
    intact); payload verification is the separate, explicit `verify()` call so large
    bundles can be opened cheaply for inspection.
    """
    
    def __init__(
            self,
            root: Union[str, Path],  # Extracted bundle root (contains bundle.json)
        )
    
    def from_zip(
            cls,
            zip_path: Union[str, Path],                     # Bundle .zip
            extract_to: Optional[Union[str, Path]] = None,  # Destination (default: fresh temp dir)
        ) -> "BundleReader":  # Reader over the extracted directory
        "Extract (guarded) + open."
    
    def all_entries(self) -> List[PayloadEntry]:  # Every payload entry in the bundle
            """Flatten stage manifests + stage artifacts + attachments."""
            out: List[PayloadEntry] = []
        "Flatten stage manifests + stage artifacts + attachments."
    
    def resolve(
            self,
            entry: PayloadEntry,  # Any payload entry
        ) -> Optional[Path]:  # Local path candidate (unverified for externals), or None
        "Resolve through the relocation map: copied -> bundle path; referenced ->
binding or the original location if it still exists. Never consults the wrapped
manifests' internal absolute paths."
    
    def bind(
        "Bind a local file to a referenced payload; hash-verifies before accepting."
    
    def pending_externals(self) -> List[PayloadEntry]:  # Referenced entries with no local candidate
            """Referenced payloads that currently resolve nowhere (bind to verify later)."""
            return [e for e in self.all_entries()
                    if e.disposition == DISPOSITION_REFERENCED and self.resolve(e) is None]
    
        def verify(
            self,
            check_external: bool = True,  # Also attempt referenced payloads
        ) -> VerificationReport:  # Full report
        "Referenced payloads that currently resolve nowhere (bind to verify later)."
    
    def verify(
            self,
            check_external: bool = True,  # Also attempt referenced payloads
        ) -> VerificationReport:  # Full report
        "Verify chain + payloads (three-way external honesty), honoring bindings."
    
    def stage_manifest_dict(
            self,
            stage: StageRecord,  # A chain stage
        ) -> dict:  # The wrapped run manifest, parsed
        "Load a stage's wrapped run manifest from inside the bundle."

staging (staging.ipynb)

Export-side payload staging + atomic packaging: stage files into a working directory (copied verbatim – hash-stable), then finalize as bundle.json + sidecar + ZIP via temp-name + fsync + atomic rename. An interrupted export leaves only a .partial-* temp file; no half-bundle can ever be taken for a bundle (ratified stress item 7 by construction).

Import

from cjm_provenance_bundle.staging import (
    safe_component,
    BundleStaging
)

Functions

def safe_component(
    name: str,  # Arbitrary filename component
) -> str:  # Filesystem/zip-safe component
    "Sanitize a path component for use inside the bundle archive."

Classes

class BundleStaging:
    def __init__(
        self,
        work_dir: Optional[Union[str, Path]] = None,  # Working dir (default: fresh temp dir)
    )
    """
    Accumulates payloads into a working directory, then finalizes atomically.
    
    `add_file` hashes exactly once and enforces export-time honesty: when a
    `known_hash` is supplied (e.g. the run manifest's recorded source hash) and the
    local file disagrees, staging fails LOUDLY -- the source drifted since the run.
    """
    
    def __init__(
            self,
            work_dir: Optional[Union[str, Path]] = None,  # Working dir (default: fresh temp dir)
        )
    
    def add_file(
            self,
            src: Union[str, Path],              # Source file on the exporting machine
            payload_class: str,                 # Semantic payload class
            disposition: str,                   # DISPOSITION_COPIED | DISPOSITION_REFERENCED
            arcdir: str,                        # Bundle-relative directory ("stages/<id>" / "attachments")
            original: Optional[Dict[str, Any]] = None,  # Locator dict override (default: FileRef(src))
            known_hash: Optional[str] = None,   # Recorded hash to honor (e.g. from the run manifest)
        ) -> PayloadEntry:  # The staged entry
        "Stage one payload; returns its entry with hash + (for copied) bundle path."
    
    def finalize(
            self,
            bundle: BundleManifest,        # The completed envelope manifest
            out_path: Union[str, Path],    # Destination .zip path
        ) -> Path:  # The written bundle path
        "Write bundle.json + sidecar, pack to a temp zip, fsync, atomic-rename."
    
    def abort(self) -> None
        "Discard the working directory (export failed; leave nothing behind)."