Databricks: bind a committed table version to exact handoff bytes.

Where Makoto sits

The job has three tasks. Only the first signs; the second verifies; the third depends on the second, so the job’s own dependency rules enforce the gate.

1. export_and_attestReads one Delta version, writes a single CSV to a Unity Catalog Volume, signs it with the version as source metadata, builds the handoff, and passes the run ID and anchors as task values.
2. makoto_gateVerifies the bundle. MakotoDenied is not caught, so the task fails.
3. publishDepends on makoto_gate; with the default run_if of all succeeded, it does not run after a denial.

1. Put the verifier on the cluster

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 makoto_boundary.py in src/ beside the task scripts.

An init script installs uv and the pinned core on every node of the job cluster.

#!/bin/bash
# Cluster init script: the pinned reference verifier on every node.
set -euo pipefail
curl -LsSf https://astral.sh/uv/install.sh | env UV_INSTALL_DIR=/usr/local/bin sh
git clone https://github.com/makoto-project/makoto.git /opt/makoto
git -C /opt/makoto checkout 276008eeec0f10d5322c4c627b8c103222fee841
cd /opt/makoto && uv sync --locked --dev

uv run makoto_bootstrap.py /Volumes/main/finance/makoto \
  --source-kind urn:example:source:delta-table-version \
  --source-uri databricks://acme/main.sales.orders \
  --operation-type urn:example:operation:none
export MAKOTO_KEYS=/Volumes/main/finance/makoto/keys MAKOTO_POLICY=/Volumes/main/finance/makoto/policy.json

2. The job definition

A Databricks Asset Bundle, databricks.yml:

bundle:
  name: makoto-orders

resources:
  jobs:
    makoto_orders:
      name: makoto-orders
      job_clusters:
        - job_cluster_key: main
          new_cluster:
            spark_version: 15.4.x-scala2.12
            node_type_id: i3.xlarge
            num_workers: 1
            spark_env_vars:
              MAKOTO_CORE: /opt/makoto
              MAKOTO_KEYS: /Volumes/main/sales/makoto_keys
              MAKOTO_POLICY: /Volumes/main/finance/makoto/policy.json
            init_scripts:
              # Clones the pinned core revision and runs uv sync on every node.
              - workspace:
                  destination: /Workspace/Shared/makoto/install-makoto.sh
      tasks:
        - task_key: export_and_attest
          job_cluster_key: main
          spark_python_task:
            python_file: ./src/export_and_attest.py
        - task_key: makoto_gate
          depends_on:
            - task_key: export_and_attest
          job_cluster_key: main
          spark_python_task:
            python_file: ./src/makoto_gate.py
        - task_key: publish
          depends_on:
            - task_key: makoto_gate
          job_cluster_key: main
          spark_python_task:
            python_file: ./src/publish.py

3. The tasks

# src/export_and_attest.py
"""Task 1: export a pinned Delta version to a Volume and sign the exact file."""

import json
import os
import uuid
from pathlib import Path

from databricks.sdk.runtime import dbutils, spark

import makoto_boundary as mb

TABLE = "main.sales.orders"
HANDOFFS = Path(os.environ.get("MAKOTO_HANDOFFS", "/Volumes/main/sales/handoffs"))
RECIPIENT = "example:finance-workspace"

run = uuid.uuid4().hex
event = lambda name: f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, f'{run}/{name}')}"

# Read one committed table version, never "latest" twice.
version = spark.sql(f"DESCRIBE HISTORY {TABLE} LIMIT 1").first()["version"]
orders = spark.read.option("versionAsOf", version).table(TABLE)

out_dir = HANDOFFS / run / "orders"
(orders.orderBy("order_id").coalesce(1)
    .write.mode("overwrite").option("header", True).csv(str(out_dir)))
if not (out_dir / "_SUCCESS").exists():
    raise RuntimeError("export was not committed")
(export,) = sorted(out_dir.glob("part-*.csv"))

head = mb.sign_origin(
    export, HANDOFFS / run / "attestations" / "01-export.dsse.json",
    source_kind="urn:example:source:delta-table-version",
    source_uri=f"databricks://acme/{TABLE}",
    source_metadata={"version": str(version), "mediaType": "text/csv"},
    event_id=event("export"),
)
anchors = mb.create_handoff(
    head, export, HANDOFFS / run / "attestations", HANDOFFS / run / "bundle",
    recipient=RECIPIENT, bundle_id=event("bundle"), media_type="text/csv",
)
# Task values carry small strings between tasks. Keys and bundles never travel here.
dbutils.jobs.taskValues.set(key="run", value=run)
dbutils.jobs.taskValues.set(key="anchors", value=json.dumps(anchors))

# src/makoto_gate.py
"""Task 2: verify the handoff. Raising fails this task, so publish never runs."""

import json
import os
from pathlib import Path

from databricks.sdk.runtime import dbutils

import makoto_boundary as mb

HANDOFFS = Path(os.environ.get("MAKOTO_HANDOFFS", "/Volumes/main/sales/handoffs"))
RECIPIENT = "example:finance-workspace"

run = dbutils.jobs.taskValues.get(taskKey="export_and_attest", key="run", debugValue="")
anchors = json.loads(dbutils.jobs.taskValues.get(taskKey="export_and_attest", key="anchors", debugValue="{}"))

report = mb.verify(
    HANDOFFS / run / "bundle", anchors,
    recipient=RECIPIENT, report_path=HANDOFFS / run / "verification.json",
)
print(f"Makoto {report['decision']} for {run}, policy {report['policyDigest']['sha256'][:12]}")

# src/publish.py
"""Task 3: runs only after makoto_gate succeeded."""

from databricks.sdk.runtime import dbutils

run = dbutils.jobs.taskValues.get(taskKey="export_and_attest", key="run", debugValue="")
print(f"publishing handoff {run}")

4. Deploy and run

databricks bundle validate
databricks bundle deploy
databricks bundle run makoto_orders

5. What a gate failure looks like

From the local run of the gate task after the delivered file was changed; the task exits 1, which marks it failed:

makoto_boundary.MakotoDenied: Makoto deny (E_ARTIFACT_DIGEST): step 12 E_ARTIFACT_DIGEST: final artifact bytes differ from the signed digest

Design notes

Why read by version

Two reads of main.sales.orders during one job can see different commits. versionAsOf pins the export to the version recorded in the statement.

Keys

Store signing keys in a Volume or secret scope readable only by the job’s service principal, one key per role. Task values carry run IDs and digests, never keys.

Production-sized tables

Do not coalesce to one file. Sign a dataset manifest over the part files, as described on the Spark page.

Other attachment points on this platform

Delta commit

After a write commits, record table identity, version, job, cluster, and notebook context. Resolve the exact snapshot files or a deterministic manifest.

Feature or model export

Attest the stable exported artifact; reference the upstream table statement and exact input digest.

Unity Catalog

Store pointers or searchable metadata if useful. Statement and artifact digests remain the portable identities.

External handoff

Sign the exact roots, heads, profiles, and final artifacts given to the recipient; keep receiver policy outside the bundle.

What the evidence can say

It can bind a Databricks commit or export to exact predecessor and output evidence. It cannot prove that every notebook cell ran, that a quality expectation was correct, or that a model card automatically carries the data history.

Troubleshooting

ModuleNotFoundError: No module named 'makoto_boundary'

Place the helper beside the task scripts in the bundle’s src/ directory; the script directory is on the import path of a Python script task.

uv: command not found in a task

The init script did not run on that cluster, or ran on a different cluster than the task. Confirm the task uses the job cluster that declares it.