Prefect: attest after persistence, not after a Python return.

Where Makoto sits

Each producing task returns a path to bytes it has already written. A separate signing task turns that path into evidence, and a retry_condition_fn separates transient tool failures from verdicts.

1. extract → attest_originWrite the raw export, then sign an origin statement over its bytes.
2. normalize → attest_normalizeWrite the output, then sign a transformation bound to the raw bytes and origin statement.
3. create_handoff → deliverBuild the bundle and move it to the receiver; a flow parameter can corrupt it on the way.
4. makoto_gateVerifies. MakotoDenied fails the task; retry_unless_denied declines to retry it.
5. publishTakes the gate’s result as an input, so it cannot run after a failed gate.

1. Install

Every platform on this site uses the same two files from the integration index: makoto_boundary.py, a standard-library helper that signs, hands off, and verifies through the pinned CLI, and makoto_bootstrap.py, which creates one key per claim role and the receiver’s trust policy. Put both next to the code below.

git clone https://github.com/makoto-project/makoto.git /opt/makoto
git -C /opt/makoto checkout 276008eeec0f10d5322c4c627b8c103222fee841
(cd /opt/makoto && uv sync --locked --dev)
export MAKOTO_CORE=/opt/makoto
uv pip install "prefect==3.8.6"

uv run makoto_bootstrap.py /srv/makoto \
  --source-kind urn:example:source:orders-export \
  --source-uri urn:example:orders:daily \
  --operation-type urn:example:operation:normalize-orders
export MAKOTO_KEYS=/srv/makoto/keys MAKOTO_POLICY=/srv/makoto/policy.json

2. The flow

Save this as orders_flow.py.

"""A Prefect flow that signs each task output and refuses a handoff Makoto denies."""

import os
import shutil
import sys
import uuid
from pathlib import Path

from prefect import flow, get_run_logger, task
from prefect.runtime import flow_run

import makoto_boundary as mb

WORK = Path(os.environ["MAKOTO_WORK"])
RECIPIENT = "example:analytics-team"


def run_dir(*parts: str) -> Path:
    path = WORK.joinpath("runs", str(flow_run.id), *parts)
    path.mkdir(parents=True, exist_ok=True)
    return path


def event_id(name: str) -> str:
    return f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, f'{flow_run.id}/{name}')}"


def retry_unless_denied(task, task_run, state) -> bool:
    """Retry tool failures; never retry a verdict or a conflicting signature."""
    try:
        state.result()
    except (mb.MakotoDenied, mb.MakotoConflict):
        return False
    except Exception:
        return True
    return False


@task
def extract() -> Path:
    raw = run_dir("data") / "orders.raw.csv"
    raw.write_text("order_id,region,total\n1001, us ,42.00\n1002,GB,17.50\n")
    return raw


@task(retries=2, retry_condition_fn=retry_unless_denied)
def attest_origin(raw: Path) -> Path:
    return mb.sign_origin(
        raw, run_dir("attestations") / "01-origin.dsse.json",
        source_kind="urn:example:source:orders-export",
        source_uri="urn:example:orders:daily",
        event_id=event_id("origin"),
    )


@task
def normalize(raw: Path) -> Path:
    header, *rows = raw.read_text().splitlines()
    out = run_dir("data") / "orders.normalized.csv"
    out.write_text("\n".join([header, *(",".join(c.strip().lower() for c in r.split(",")) for r in rows)]) + "\n")
    return out


@task(retries=2, retry_condition_fn=retry_unless_denied)
def attest_normalize(raw: Path, origin: Path, normalized: Path) -> Path:
    return mb.sign_transform(
        normalized, run_dir("attestations") / "02-normalize.dsse.json",
        inputs=[(raw, origin)],
        operation_type="urn:example:operation:normalize-orders",
        event_id=event_id("normalize"),
    )


@task(retries=2, retry_condition_fn=retry_unless_denied)
def create_handoff(head: Path, normalized: Path) -> dict:
    anchors = mb.create_handoff(
        head, normalized, run_dir("attestations"), run_dir() / "bundle",
        recipient=RECIPIENT, bundle_id=event_id("bundle"), media_type="text/csv",
    )
    return {"bundle": str(run_dir() / "bundle"), "anchors": anchors}


@task
def deliver(handoff: dict, tamper_in_transit: bool) -> Path:
    delivered = run_dir("receiver") / "bundle"
    shutil.copytree(handoff["bundle"], delivered, dirs_exist_ok=True)
    if tamper_in_transit:
        final = next((delivered / "artifacts" / "final").glob("*.bin"))
        final.write_bytes(final.read_bytes() + b"1003,us,0.01\n")
    return delivered


@task(retries=2, retry_condition_fn=retry_unless_denied)
def makoto_gate(delivered: Path, handoff: dict) -> dict:
    report = mb.verify(
        delivered, handoff["anchors"],
        recipient=RECIPIENT, report_path=run_dir("receiver") / "verification.json",
    )
    return {"decision": report["decision"], "policy": report["policyDigest"]["sha256"]}


@task
def publish(normalized: Path, verdict: dict) -> None:
    get_run_logger().info(
        "publishing %s under Makoto %s (policy %s)", normalized.name, verdict["decision"], verdict["policy"][:12]
    )


@flow(name="makoto-orders")
def makoto_orders(tamper_in_transit: bool = False) -> None:
    raw = extract()
    origin = attest_origin(raw)
    normalized = normalize(raw)
    head = attest_normalize(raw, origin, normalized)
    handoff = create_handoff(head, normalized)
    verdict = makoto_gate(deliver(handoff, tamper_in_transit), handoff)
    publish(normalized, verdict)


if __name__ == "__main__":
    makoto_orders(tamper_in_transit="--tamper-in-transit" in sys.argv)

3. Run it

export MAKOTO_WORK=/srv/makoto/work
python orders_flow.py

Flow log, timestamps and task-run suffixes trimmed:

| INFO    | Flow run - Beginning flow run for flow 'makoto-orders'
| INFO    | Task run 'extract' - Finished in state Completed()
| INFO    | Task run 'attest_origin' - Finished in state Completed()
| INFO    | Task run 'normalize' - Finished in state Completed()
| INFO    | Task run 'attest_normalize' - Finished in state Completed()
| INFO    | Task run 'create_handoff' - Finished in state Completed()
| INFO    | Task run 'deliver' - Finished in state Completed()
| INFO    | Task run 'makoto_gate' - Finished in state Completed()
| INFO    | Task run 'publish' - publishing orders.normalized.csv under Makoto allow (policy 4bd642394c20)
| INFO    | Task run 'publish' - Finished in state Completed()
| INFO    | Flow run - Finished in state Completed()

4. What a gate failure looks like

python orders_flow.py --tamper-in-transit

| INFO    | Flow run - Beginning flow run for flow 'makoto-orders'
| INFO    | Task run 'extract' - Finished in state Completed()
| INFO    | Task run 'attest_origin' - Finished in state Completed()
| INFO    | Task run 'normalize' - Finished in state Completed()
| INFO    | Task run 'attest_normalize' - Finished in state Completed()
| INFO    | Task run 'create_handoff' - Finished in state Completed()
| INFO    | Task run 'deliver' - Finished in state Completed()
| ERROR   | Task run 'makoto_gate' - Finished in state Failed('Task run encountered an exception MakotoDenied: Makoto deny (E_ARTIFACT_DIGEST): step 12 E_ARTIFACT_DIGEST: final artifact bytes differ from the signed digest')
| INFO    | Flow run - Finished in state Failed('Flow run encountered an exception: MakotoDenied: Makoto deny (E_ARTIFACT_DIGEST): step 12 E_ARTIFACT_DIGEST: final artifact bytes differ from the signed digest')

makoto_gate fails on its first attempt with no retry, publish never starts, and the flow run ends Failed with the primary error in its message.

Retries that respect a verdict

Prefect calls retry_condition_fn with the failed state. Re-raising the state’s exception is how the function sees what went wrong. A MakotoDenied or MakotoConflict will be the same on every attempt, so the function returns False; anything else, including MakotoToolError, is retried.

def retry_unless_denied(task, task_run, state) -> bool:
    """Retry tool failures; never retry a verdict or a conflicting signature."""
    try:
        state.result()
    except (mb.MakotoDenied, mb.MakotoConflict):
        return False
    except Exception:
        return True
    return False

Other attachment points on this platform

Task completion

Use a post-task hook only after the task has durably committed an artifact whose exact identity can be resolved.

Flow handoff

Create a signed handoff when artifacts cross a team, account, region, or external recipient boundary.

Retries and cache

Make attestation idempotent; distinguish a reused cached artifact from a newly produced artifact.

Prefect context

Carry flow run, task run, deployment, work pool, parameters, and code revision as operation metadata.

What the evidence can say

It can bind a durable task output to exact inputs and selected Prefect context. It cannot treat an arbitrary return value as an artifact or infer an authorized organizational handoff from flow nesting.

Troubleshooting

A denied gate retries twice before failing

The task needs retry_condition_fn=retry_unless_denied in addition to retries. Without it, Prefect retries every exception.

Cached task results skip signing

If you enable result caching on a producing task, a cache hit returns an old path without rewriting it. Key the cache on the input digest, and let the signing task reuse the existing envelope, which makoto_boundary does when the bytes match.