Customize the delivery manifest

load("@aspect//traits.axl", "DeliveryTrait")

def _runfiles_path(entry, suffix):
    """`<runfiles_workspace_dir>/<package>/<target_name><suffix>` for a local label."""
    rwd = entry.get("runfiles_workspace_dir")
    if not rwd:
        return None
    label = entry["label"]
    if not label.startswith("//"):
        return None  # external repo deliverables need a different workspace path
    package, _, target_name = label.removeprefix("//").rpartition(":")
    if package:
        return "{}/{}/{}{}".format(rwd, package, target_name, suffix)
    return "{}/{}{}".format(rwd, target_name, suffix)

def _on_delivery_target(entry):
    if entry["outcome"] != "ok":
        return None
    digest_path = _runfiles_path(entry, suffix = ".digest")
    if not digest_path or not ctx.std.fs.exists(digest_path):
        return None
    return {
        "image_digest": ctx.std.fs.read_to_string(digest_path).strip(),
    }

def config(ctx: ConfigContext):
    ctx.traits[DeliveryTrait].delivery_target = _on_delivery_target

def _on_delivery_target(entry):
    if entry["outcome"] != "ok":
        return None
    for path in entry.get("default_outputs", []):
        if path.endswith(".layer"):
            return {"layer_path": path}
    return None

def _on_delivery_target(entry):
    if entry["outcome"] != "ok":
        return None
    extras = {}

    digest_path = _runfiles_path(entry, suffix = ".digest")
    if digest_path and ctx.std.fs.exists(digest_path):
        extras["image_digest"] = ctx.std.fs.read_to_string(digest_path).strip()

    for path in entry.get("default_outputs", []):
        if path.endswith(".layer"):
            extras["layer_path"] = path
            break

    return extras if extras else None

_MANIFEST_YAML = """\
deliveries:
{% for d in deliveries %}
  - label: {{ d.label }}
    outcome: {{ d.outcome }}
    {% if d.custom.image_digest is defined %}image_digest: {{ d.custom.image_digest }}{% endif %}
{% endfor %}
"""

def _render_manifest_file(manifest):
    return ctx.template.jinja2(_MANIFEST_YAML, data = manifest)

def config(ctx: ConfigContext):
    ctx.traits[DeliveryTrait].render_manifest_file = _render_manifest_file

def _upload_manifest(manifest):
    # Disable uploads on a recognized CI host that already exposes the
    # manifest via its own surface (e.g. an in-house dashboard).
    return []

def config(ctx: ConfigContext):
    ctx.traits[DeliveryTrait].upload_manifest = _upload_manifest

def _upload_manifest(manifest):
    rendered = ctx.template.jinja2(_MANIFEST_YAML, data = manifest)
    return [{"name": "delivery-manifest.yml", "content": rendered}]

def config(ctx: ConfigContext):
    ctx.traits[DeliveryTrait].upload_manifest = _upload_manifest

_DELIVERY_IMAGES_YAML = """\
images:
{% for d in deliveries %}{% if d.outcome == "ok" and d.custom.image_digest is defined %}
  - label: {{ d.label }}
    image_digest: {{ d.custom.image_digest }}
    {% if d.custom.layer_path is defined %}layer_path: {{ d.custom.layer_path }}{% endif %}
{% endif %}{% endfor %}
"""

def _upload_manifest(manifest):
    return [
        {
            "name": "delivery-manifest.json",
            "content": json.encode(manifest, indent = 2),
        },
        {
            "name": "delivery-images.yml",
            "content": ctx.template.jinja2(_DELIVERY_IMAGES_YAML, data = manifest),
        },
    ]

def config(ctx: ConfigContext):
    ctx.traits[DeliveryTrait].upload_manifest = _upload_manifest

def _upload_manifest(manifest):
    # Ship JSON for tooling, YAML for human review, and a CSV of the
    # successful deliveries for the release-notes pipeline.
    successes = [d for d in manifest["deliveries"] if d["outcome"] == "ok"]
    csv = ["label,output_sha"] + [
        "{},{}".format(d["label"], d["output_sha"]) for d in successes
    ]
    return [
        {"name": "delivery-manifest.json", "content": json.encode(manifest, indent = 2)},
        {"name": "delivery-manifest.yml", "content": ctx.template.jinja2(_MANIFEST_YAML, data = manifest)},
        {"name": "successes.csv", "content": "\n".join(csv) + "\n"},
    ]

def config(ctx: ConfigContext):
    ctx.traits[DeliveryTrait].upload_manifest = _upload_manifest

def _on_delivery_manifest(manifest):
    """Post the manifest to an escrow registry once all targets are done."""
    if not manifest["deliveries"]:
        return  # early-exit path (validation / build failure) — nothing to post

    digests = [
        d["custom"]["image_digest"]
        for d in manifest["deliveries"]
        if d["outcome"] == "ok" and "image_digest" in d.get("custom", {})
    ]
    if not digests:
        return

    print("[delivery_manifest] posting {} digest(s) to escrow".format(len(digests)))
    # ... your in-process post / OCI layer assembly / audit-log write here ...

def config(ctx: ConfigContext):
    ctx.traits[DeliveryTrait].delivery_manifest = _on_delivery_manifest