From bc86a0875b3a2119db9e4f72b61d58e0845b2664 Mon Sep 17 00:00:00 2001
From: Dov Alperin <dov.alperin@materialize.com>
Date: Wed, 6 May 2026 19:04:13 -0400
Subject: [PATCH] antithesis: scaffold harness, multi-scenario layout,
 broadened testdrive corpus
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Brings up the Antithesis Test Composer harness for Materialize. The branch
covers research artifacts, the test-driver mzbuild image, scenario
infrastructure, workload helpers (testdrive runner, corpus lists, the
upsert-sources prototype), and a few small upstream fixes that were
prerequisites.

Layout
------

    antithesis/
      AGENTS.md                          directory map + scenarios table
      scratchbook/                       SUT analysis, deployment topology,
                                         test-driver integration plan,
                                         scenario strategy, existing-assertions
                                         inventory
      configs/<scenario>/
        mzcompose.py                     source-of-truth composition
        docker-compose.yaml              generated artifact (snouty consumes)
      bin/render-compose-yaml.py         renders configs/<scenario>/docker-compose.yaml
                                         from mzcompose.py and layers on
                                         platform/hostname/container_name/
                                         NO_COLOR Antithesis attributes
      test-driver/                       mzbuild image: MZFROM testdrive +
                                         Python + Antithesis SDK + the workload
                                         tree at /opt/antithesis/test/v1/ +
                                         curated test/testdrive corpus at
                                         /opt/materialize/td/testdrive/
      test/v1/
        helper_bootstrap.py              shared sys.path injector
        testdrive_{sql,kafka,load_generator,recovery}/
                                         area templates; each picks a random
                                         .td from materialize.antithesis.testdrive_corpus
        upsert_sources/                  randomized helper-driven upsert
                                         workload with expected-state model

    misc/python/materialize/antithesis/
      sdk.py                             SDK wrapper with local fallbacks; the
                                         antithesis package coexists with our
                                         scaffolding directory because both
                                         are namespace packages
      testdrive_config.py                shared TestdriveConfig dataclass
      td_runner.py                       generic .td runner: subprocess +
                                         tolerated-failure retry + reachable()
                                         on success and on tolerated failure
      testdrive_corpus.py                curated lists of base-compatible .td
                                         files split into 4 area buckets
                                         (BASE_SQL=22, BASE_KAFKA=35,
                                         BASE_LOAD_GENERATOR=12,
                                         BASE_RECOVERY=8 — 72 unique files)
      upsert_sources.py                  prototype workload helper

Why per-scenario configs
------------------------

`antithesis/scratchbook/scenario-strategy.md` documents the design with
citations to the Antithesis docs (snouty docs CLI). In short: incompatible
topologies (e.g. base SQL+Kafka vs MySQL CDC with multithreaded replicas)
get separate config dirs and separate `snouty launch` invocations;
incompatible workloads on the same topology coexist as multiple test
templates inside one config (Antithesis selects exactly one template per
execution history). The base scenario is the only one wired up here;
`mysql_mt_replicas` for the SS-95 ticket is the planned next addition.

Why no per-area eventually_* recovery checks
--------------------------------------------

Earlier drafts of this branch had per-area `eventually_*` commands. They
either reduced to tautologies (`SELECT count(*) >= 0` always passes if
pgwire is up) or to a generic CREATE/INSERT/SELECT/DROP round-trip that
was only weakly correlated with the chosen .td's actual semantics. When
the singleton picks a random .td you can't write a useful recovery
property without knowing what state was created. Real recovery
properties belong either in SUT-side Rust assertions or in
scenario-specific `eventually_*` commands tied to a specific workload —
which is exactly the shape `upsert_sources/eventually_*` already has
(it writes a sentinel and waits for it). The scratchbook entry warns
against re-adding generic per-area recovery checks.

Upstream fixes pulled in (could be split out later)
---------------------------------------------------

* misc/python/materialize/cli/mzcompose.py — the shtab-Enum-choices
  workaround broke `--arch` and `--sanitizer` on Python 3.13. Argparse's
  post-conversion `member in choices` check failed because choices were
  member names while `type=Enum` returned member objects. Switched to
  `list(action.choices)` (Enum members) so argparse and shtab are both
  happy. Includes the regenerated bash/zsh shell completions.

* misc/python/materialize/mzbuild.py — the Copy pre-image plugin was
  unused upstream and crashed when used: `Copy.inputs()` returned paths
  relative to the source dir, but the mzbuild fingerprinter expected
  paths relative to the repo root, so `os.lstat(rd.root / rel_path)` hit
  FileNotFoundError. Made `Copy.inputs()` repo-root-relative and
  `Copy.run()` strip the source prefix before computing the destination
  inside the build context.

* misc/python/materialize/mzcompose/service.py — added `container_name`
  to `ServiceConfig` (a real Compose field). Antithesis requires it for
  log/fault attribution.

* ci/test/lint-main/checks/check-mzcompose-files.sh — exclude
  `antithesis/configs/*/mzcompose.py` from the 'unused in any CI
  pipeline file' check; Antithesis runs are submitted via snouty, not
  Buildkite.

* ci/builder/requirements.txt — added `antithesis==0.2.0` so the SDK
  resolves locally for type-checking and ad-hoc imports. Coexists with
  our `antithesis/` scaffolding directory because both are namespace
  packages and the merge picks up submodules from site-packages.

Known limitations
-----------------

* On Apple Silicon, `snouty validate antithesis/configs/base` fails
  end-to-end because the amd64 `materialized` image's `clusterd` child
  segfaults under Rosetta during lgalloc init
  (`unix_wait_status(11)` -> container `Exited (139)`). Run validate on
  Linux/x86 instead. Documented in antithesis/AGENTS.md.

* On Apple Silicon, `bin/mzimage acquire --arch x86_64` currently fails
  to link with `ld.lld: error: undefined symbol: getauxval`: the
  `materializeinc/crosstools/x86_64-unknown-linux-gnu` homebrew formula
  ships glibc 2.12.1, but Rust 1.95's stdlib references getauxval which
  needs glibc 2.16+. Workaround:
  `bin/ci-builder run stable bin/mzimage acquire --arch x86_64 antithesis-test-driver`
  uses the Docker builder's current glibc. The homebrew formula needs an
  upstream update; CI is unaffected (Linux hosts route through ci-builder
  by default).

Next steps
----------

* SS-95: `mysql_mt_replicas` scenario per scratchbook/test-driver-integration.md
* Tier 2 scenarios: pg_cdc, mysql_cdc, sql_server_cdc, s3_copy
* Tier 3 structural refactors: parallel-workload regression scenario
  (gate Database.create() Kafka/CSR/AWS/PG/MySQL/SQLServer/Iceberg
  CONNECTIONs on flags), zippy execution adapter
* Per-scenario template gating via `ANTITHESIS_SCENARIO` env in the
  test-driver entrypoint, so a scenario only sees its compatible
  templates
* SUT-side Rust assertions where they're justified (rare/dangerous
  internal states, branch outcomes)
---
 antithesis/AGENTS.md                          | 104 ++++++
 antithesis/bin/render-compose-yaml.py         | 201 ++++++++++++
 antithesis/configs/base/docker-compose.yaml   | 113 +++++++
 antithesis/configs/base/mzcompose.py          | 119 +++++++
 antithesis/scratchbook/deployment-topology.md | 112 +++++++
 antithesis/scratchbook/existing-assertions.md |  64 ++++
 antithesis/scratchbook/scenario-strategy.md   | 101 ++++++
 antithesis/scratchbook/sut-analysis.md        | 106 ++++++
 .../scratchbook/test-driver-integration.md    | 301 ++++++++++++++++++
 antithesis/setup-complete.sh                  |  32 ++
 antithesis/test-driver/.gitignore             |   5 +
 antithesis/test-driver/Dockerfile             |  91 ++++++
 antithesis/test-driver/entrypoint.sh          | 105 ++++++
 antithesis/test-driver/mzbuild.yml            |  41 +++
 antithesis/test/v1/helper_bootstrap.py        |  20 ++
 .../singleton_driver_random_kafka             |  32 ++
 .../singleton_driver_random_load_generator    |  33 ++
 .../singleton_driver_random_recovery          |  39 +++
 .../testdrive_sql/singleton_driver_random_sql |  34 ++
 .../anytime_upsert_source_health              |  17 +
 .../eventually_upsert_source_catches_up       |  17 +
 .../finally_upsert_expected_rows_visible      |  19 ++
 .../upsert_sources/first_create_upsert_source |  17 +
 .../parallel_driver_read_stale_safe           |  17 +
 .../parallel_driver_write_upserts             |  17 +
 ci/builder/requirements.txt                   |   6 +
 .../lint-main/checks/check-mzcompose-files.sh |   1 +
 misc/completions/bash/_mzcompose              |   2 +-
 misc/completions/zsh/_mzcompose               |   2 +-
 .../python/materialize/antithesis/__init__.py |  10 +
 misc/python/materialize/antithesis/sdk.py     |  83 +++++
 .../materialize/antithesis/td_runner.py       | 109 +++++++
 .../antithesis/testdrive_config.py            |  95 ++++++
 .../antithesis/testdrive_corpus.py            | 212 ++++++++++++
 .../materialize/antithesis/upsert_sources.py  | 266 ++++++++++++++++
 misc/python/materialize/cli/mzcompose.py      |  13 +-
 misc/python/materialize/mzbuild.py            |  19 +-
 misc/python/materialize/mzcompose/service.py  |   7 +
 38 files changed, 2574 insertions(+), 8 deletions(-)
 create mode 100644 antithesis/AGENTS.md
 create mode 100755 antithesis/bin/render-compose-yaml.py
 create mode 100644 antithesis/configs/base/docker-compose.yaml
 create mode 100644 antithesis/configs/base/mzcompose.py
 create mode 100644 antithesis/scratchbook/deployment-topology.md
 create mode 100644 antithesis/scratchbook/existing-assertions.md
 create mode 100644 antithesis/scratchbook/scenario-strategy.md
 create mode 100644 antithesis/scratchbook/sut-analysis.md
 create mode 100644 antithesis/scratchbook/test-driver-integration.md
 create mode 100755 antithesis/setup-complete.sh
 create mode 100644 antithesis/test-driver/.gitignore
 create mode 100644 antithesis/test-driver/Dockerfile
 create mode 100755 antithesis/test-driver/entrypoint.sh
 create mode 100644 antithesis/test-driver/mzbuild.yml
 create mode 100755 antithesis/test/v1/helper_bootstrap.py
 create mode 100755 antithesis/test/v1/testdrive_kafka/singleton_driver_random_kafka
 create mode 100755 antithesis/test/v1/testdrive_load_generator/singleton_driver_random_load_generator
 create mode 100755 antithesis/test/v1/testdrive_recovery/singleton_driver_random_recovery
 create mode 100755 antithesis/test/v1/testdrive_sql/singleton_driver_random_sql
 create mode 100755 antithesis/test/v1/upsert_sources/anytime_upsert_source_health
 create mode 100755 antithesis/test/v1/upsert_sources/eventually_upsert_source_catches_up
 create mode 100755 antithesis/test/v1/upsert_sources/finally_upsert_expected_rows_visible
 create mode 100755 antithesis/test/v1/upsert_sources/first_create_upsert_source
 create mode 100755 antithesis/test/v1/upsert_sources/parallel_driver_read_stale_safe
 create mode 100755 antithesis/test/v1/upsert_sources/parallel_driver_write_upserts
 create mode 100644 misc/python/materialize/antithesis/__init__.py
 create mode 100644 misc/python/materialize/antithesis/sdk.py
 create mode 100644 misc/python/materialize/antithesis/td_runner.py
 create mode 100644 misc/python/materialize/antithesis/testdrive_config.py
 create mode 100644 misc/python/materialize/antithesis/testdrive_corpus.py
 create mode 100644 misc/python/materialize/antithesis/upsert_sources.py

diff --git a/antithesis/AGENTS.md b/antithesis/AGENTS.md
new file mode 100644
index 0000000000000..62574c2604f73
--- /dev/null
+++ b/antithesis/AGENTS.md
@@ -0,0 +1,104 @@
+This directory contains files relevant to running tests in Antithesis.
+
+Use the `antithesis-setup` skill to scaffold and manage this directory. Use
+the `antithesis-research` skill to analyze the system and build a property
+catalog. Use the `antithesis-workload` skill to implement assertions and test
+commands. Use the `antithesis-launch` skill to build, validate, and submit
+Antithesis runs — do not run `snouty run` directly.
+
+## Scenarios
+
+Antithesis runs are organized as **N narrow scenarios**, each with its own
+compose file under `configs/<name>/`. The split rule and rationale live in
+`scratchbook/scenario-strategy.md`. Each scenario is launched separately via
+`snouty launch -c antithesis/configs/<name> --source <name>`.
+
+Available scenarios:
+
+| Scenario | Topology | Status |
+| --- | --- | --- |
+| `base` | materialized + redpanda + driver | active |
+
+When adding a scenario:
+
+1. Drop a new `configs/<name>/mzcompose.py` defining its `SERVICES`.
+2. Run `bin/pyactivate antithesis/bin/render-compose-yaml.py --scenario <name>`
+   to produce `configs/<name>/docker-compose.yaml`.
+3. Add an entry to the table above.
+
+## snouty validate
+
+Validate a scenario locally:
+
+    snouty validate antithesis/configs/<scenario>
+
+The validator runs Docker Compose, watches for the `setup_complete` event,
+and inspects discovered Test Composer commands. Run `bin/mzcompose --find
+<scenario> build` (or `bin/mzimage acquire --arch x86_64 antithesis-test-driver`)
+first to make sure the local images are present.
+
+> **Apple Silicon caveat.** The compose configs pin every service to
+> `platform: linux/amd64` because the Antithesis platform is amd64-only. On
+> macOS / Apple Silicon, Docker runs amd64 containers under Rosetta, and the
+> `clusterd` subprocess that `materialized` spawns segfaults during lgalloc
+> initialization (`ExitStatus(unix_wait_status(11))` → container `Exited (139)`).
+> This is a host-side emulation limitation, not a harness bug. Run
+> `snouty validate` on Linux/amd64 (CI, an x86 dev box, or a remote runner)
+> to exercise the full bring-up.
+
+## snouty launch
+
+Once validation passes, use the `antithesis-launch` skill to submit a run.
+
+## Directory layout
+
+**configs/**
+One subdirectory per scenario. Each contains:
+* `mzcompose.py` — source-of-truth composition for the scenario.
+* `docker-compose.yaml` — generated artifact consumed by snouty / Antithesis.
+
+**bin/**
+Helpers that operate on this directory:
+
+* `render-compose-yaml.py` — regenerates `configs/<scenario>/docker-compose.yaml`
+  from `configs/<scenario>/mzcompose.py` while layering in the
+  Antithesis-required attributes mzcompose does not emit
+  (`platform: linux/amd64`, matching `container_name`/`hostname`,
+  `NO_COLOR=1`, no host port bindings):
+
+      bin/pyactivate antithesis/bin/render-compose-yaml.py --scenario <name>
+
+**test/**
+Shared Test Composer command tree. A test template is a directory under
+`test/v1/<template>/` containing executable command files. Each command must
+have a valid prefix: `parallel_driver_`, `singleton_driver_`, `serial_driver_`,
+`first_`, `eventually_`, `finally_`, `anytime_`. Files or directories prefixed
+with `helper_` are ignored by Antithesis and can hold shared helper scripts.
+
+The `test-driver` mzbuild image bundles the entire `test/v1/` tree. Antithesis
+selects exactly one template per execution history, so a scenario's compose
+file controls **which** template runs by virtue of the topology it provides
+(e.g. the `mysql_mt_replicas` scenario would only have the MySQL templates'
+dependencies satisfied). See
+`https://antithesis.com/docs/test_templates/first_test/`.
+
+**test-driver/**
+The mzbuild image used by the `antithesis-test-driver` service. Inherits
+`MZFROM testdrive` so the testdrive binary, postgres client, and shared
+dependencies are already present. Adds Python plus the Antithesis Python
+SDK and copies in `misc/python` (so `materialize.antithesis.*` is importable)
+and `antithesis/test/v1` (so commands are discoverable at
+`/opt/antithesis/test/v1`). One image is shared across scenarios.
+
+**scratchbook/**
+Antithesis scratchbook for the codebase: SUT analysis, deployment topology,
+scenario strategy, test-driver integration plan, per-property evidence files,
+and other persistent integration notes. Keep it up to date as
+Antithesis-related decisions change.
+
+**setup-complete.sh**
+Reference fallback script that writes the `setup_complete` lifecycle event to
+`${ANTITHESIS_OUTPUT_DIR}/sdk.jsonl`. The current `test-driver` entrypoint
+emits the same event through the Antithesis Python SDK
+(`antithesis.lifecycle.setup_complete`), which is the canonical path. Keep
+this script around for any container that cannot link the SDK directly.
diff --git a/antithesis/bin/render-compose-yaml.py b/antithesis/bin/render-compose-yaml.py
new file mode 100755
index 0000000000000..9f852874f1018
--- /dev/null
+++ b/antithesis/bin/render-compose-yaml.py
@@ -0,0 +1,201 @@
+#!/usr/bin/env python3
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+"""Render `antithesis/configs/<scenario>/docker-compose.yaml` from the matching
+`antithesis/configs/<scenario>/mzcompose.py`.
+
+Why this is its own script (rather than an mzcompose workflow): `bin/mzcompose
+run <workflow>` acquires every mzbuild dependency image before invoking the
+workflow, which is wasteful when all we need is the rendered YAML. Calling
+`bin/mzcompose --find <scenario> config` directly resolves mzbuild image
+fingerprint tags without pulling.
+
+Pipeline:
+
+    1. `bin/mzcompose --find <scenario> --arch x86_64 config` produces
+       canonical compose YAML with mzbuild references resolved to ghcr.io tags
+       at the architecture Antithesis runs on (amd64).
+    2. Layer on the Antithesis-required attributes mzcompose does not emit:
+       `platform: linux/amd64`, matching `container_name` / `hostname`, and
+       `NO_COLOR=1` on every service.
+    3. Strip Docker Compose's host-port bindings. Antithesis runs hermetically;
+       host ports are noise and can collide on dev hosts.
+    4. Write `antithesis/configs/<scenario>/docker-compose.yaml`.
+
+Run via:
+
+    bin/pyactivate antithesis/bin/render-compose-yaml.py --scenario <name>
+
+`--scenario base` (the default) covers the SQL + Kafka topology. New scenarios
+live as sibling directories under `antithesis/configs/`; see
+`antithesis/scratchbook/scenario-strategy.md` for when to add one.
+"""
+
+from __future__ import annotations
+
+import argparse
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from materialize import MZ_ROOT
+
+
+def render_compose(scenario: str) -> dict[str, Any]:
+    """Run `bin/mzcompose --find <scenario> config` and return the parsed YAML.
+
+    `--arch x86_64` pins the mzbuild fingerprint to the architecture
+    Antithesis runs on. Without it, image tags would track the host arch and
+    drift between dev machines (Apple Silicon, x86 CI, …).
+    """
+    bin_mzcompose = MZ_ROOT / "bin" / "mzcompose"
+    if not bin_mzcompose.exists():
+        raise SystemExit(f"bin/mzcompose not found at {bin_mzcompose}")
+
+    raw = subprocess.run(
+        [
+            str(bin_mzcompose),
+            "--mz-quiet",
+            "--arch",
+            "x86_64",
+            "--find",
+            scenario,
+            "config",
+        ],
+        check=True,
+        capture_output=True,
+        text=True,
+    ).stdout
+    return yaml.safe_load(raw)
+
+
+def decorate_for_antithesis(compose: dict[str, Any]) -> None:
+    """Add the attributes the Antithesis platform requires that mzcompose
+    does not emit by default.
+    """
+    services = compose.get("services", {})
+    for name, svc in services.items():
+        svc.setdefault("platform", "linux/amd64")
+        svc.setdefault("container_name", name)
+        svc.setdefault("hostname", name)
+        _ensure_no_color(svc)
+        _drop_host_ports(svc)
+
+
+def _ensure_no_color(svc: dict[str, Any]) -> None:
+    """Force `NO_COLOR=1` on the service's environment block.
+
+    Antithesis stores raw bytes from container output; ANSI escape sequences
+    appear as garbage in logs and triage. The list/dict branches both reflect
+    forms that mzcompose may emit (lists with `KEY=VALUE` strings and dict
+    forms after Compose canonicalizes).
+    """
+    env = svc.get("environment")
+    if env is None:
+        svc["environment"] = ["NO_COLOR=1"]
+    elif isinstance(env, list):
+        if not any(
+            entry == "NO_COLOR=1" or entry.startswith("NO_COLOR=") for entry in env
+        ):
+            env.append("NO_COLOR=1")
+    elif isinstance(env, dict):
+        env.setdefault("NO_COLOR", "1")
+
+
+def _drop_host_ports(svc: dict[str, Any]) -> None:
+    """Remove host-side port bindings so Antithesis runs hermetically.
+
+    Containers still talk to each other across the compose network — only the
+    `host_ip`/published port is removed. mzcompose adds these for local dev
+    convenience; in Antithesis they're noise (and can collide on host).
+    """
+    ports = svc.get("ports")
+    if not ports:
+        return
+    del svc["ports"]
+
+
+_COPYRIGHT_HEADER = """\
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Generated by antithesis/bin/render-compose-yaml.py — do not hand-edit.
+# Edit antithesis/configs/<scenario>/mzcompose.py and rerun:
+#   bin/pyactivate antithesis/bin/render-compose-yaml.py --scenario <name>
+"""
+
+
+def write_compose(compose: dict[str, Any], path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    rendered = yaml.safe_dump(compose, sort_keys=False, default_flow_style=False)
+    path.write_text(_COPYRIGHT_HEADER + rendered)
+
+
+def _scenarios_dir() -> Path:
+    return MZ_ROOT / "antithesis" / "configs"
+
+
+def _available_scenarios() -> list[str]:
+    return sorted(
+        p.name
+        for p in _scenarios_dir().iterdir()
+        if p.is_dir() and (p / "mzcompose.py").exists()
+    )
+
+
+def main(argv: list[str]) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--scenario",
+        default="base",
+        help=(
+            "Antithesis scenario to render. Looks up "
+            "antithesis/configs/<scenario>/mzcompose.py and writes "
+            "antithesis/configs/<scenario>/docker-compose.yaml."
+        ),
+    )
+    parser.add_argument(
+        "--output",
+        type=Path,
+        help=(
+            "Override the default output path "
+            "(antithesis/configs/<scenario>/docker-compose.yaml)."
+        ),
+    )
+    args = parser.parse_args(argv)
+
+    scenario_dir = _scenarios_dir() / args.scenario
+    if not (scenario_dir / "mzcompose.py").exists():
+        available = ", ".join(_available_scenarios()) or "(none)"
+        raise SystemExit(
+            f"unknown scenario {args.scenario!r}: no mzcompose.py at "
+            f"{scenario_dir}. Available: {available}"
+        )
+
+    output = args.output or (scenario_dir / "docker-compose.yaml")
+
+    compose = render_compose(args.scenario)
+    decorate_for_antithesis(compose)
+    write_compose(compose, output)
+    print(f"wrote {output}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))
diff --git a/antithesis/configs/base/docker-compose.yaml b/antithesis/configs/base/docker-compose.yaml
new file mode 100644
index 0000000000000..7f00f0f60c2e5
--- /dev/null
+++ b/antithesis/configs/base/docker-compose.yaml
@@ -0,0 +1,113 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Generated by antithesis/bin/render-compose-yaml.py — do not hand-edit.
+# Edit antithesis/configs/<scenario>/mzcompose.py and rerun:
+#   bin/pyactivate antithesis/bin/render-compose-yaml.py --scenario <name>
+name: base
+services:
+  antithesis-test-driver:
+    container_name: antithesis-test-driver
+    depends_on:
+      materialized:
+        condition: service_started
+        required: true
+      redpanda:
+        condition: service_started
+        required: true
+    environment:
+      FORCE_COLOR: '0'
+      KAFKA_HOST: kafka
+      KAFKA_PORT: '9092'
+      MZ_HOST: materialized
+      MZ_HTTP_PORT: '6878'
+      MZ_KAFKA_ADDR: kafka:9092
+      MZ_MATERIALIZE_INTERNAL_URL: postgres://mz_system@materialized:6877
+      MZ_MATERIALIZE_URL: postgres://materialize@materialized:6875
+      MZ_ROOT: /opt/materialize
+      MZ_SCHEMA_REGISTRY_URL: http://schema-registry:8081
+      MZ_SQL_PORT: '6875'
+      MZ_TESTDRIVE: testdrive
+      MZ_TESTDRIVE_DEFAULT_TIMEOUT: 60s
+      NO_COLOR: '1'
+      PY_COLORS: '0'
+      PYTHONPATH: /opt/materialize/misc/python
+      SR_HOST: schema-registry
+      SR_PORT: '8081'
+      TERM: dumb
+    hostname: antithesis-test-driver
+    image: ghcr.io/materializeinc/materialize/antithesis-test-driver:mzbuild-6IXOYCUVVVFTWNHJIQWIA3LRAPKR7HGM
+    networks:
+      default: null
+    platform: linux/amd64
+    security_opt:
+    - apparmor=unconfined
+  materialized:
+    healthcheck:
+      test:
+      - CMD
+      - curl
+      - -f
+      - localhost:6878/api/readyz
+      interval: 1s
+      start_period: 10m0s
+    image: ghcr.io/materializeinc/materialize/materialized:mzbuild-VPGCGS5BPI5RKGMVZBI3QZRSAQSLCETE
+    networks:
+      default: null
+    security_opt:
+    - apparmor=unconfined
+    platform: linux/amd64
+    container_name: materialized
+    hostname: materialized
+    environment:
+    - NO_COLOR=1
+  redpanda:
+    command:
+    - redpanda
+    - start
+    - --overprovisioned
+    - --smp=1
+    - --memory=1G
+    - --reserve-memory=0M
+    - --node-id=0
+    - --check=false
+    - --set
+    - redpanda.enable_transactions=true
+    - --set
+    - redpanda.enable_idempotence=true
+    - --set
+    - redpanda.auto_create_topics_enabled=True
+    - --set
+    - redpanda.topic_memory_per_partition=4096
+    - --set
+    - --advertise-kafka-addr=kafka:9092
+    healthcheck:
+      test:
+      - CMD
+      - curl
+      - -f
+      - localhost:9644/v1/status/ready
+      interval: 1s
+      start_period: 2m0s
+    image: redpandadata/redpanda:v25.2.11
+    networks:
+      default:
+        aliases:
+        - kafka
+        - schema-registry
+    security_opt:
+    - apparmor=unconfined
+    platform: linux/amd64
+    container_name: redpanda
+    hostname: redpanda
+    environment:
+    - NO_COLOR=1
+networks:
+  default:
+    name: base_default
diff --git a/antithesis/configs/base/mzcompose.py b/antithesis/configs/base/mzcompose.py
new file mode 100644
index 0000000000000..eee5c80194f07
--- /dev/null
+++ b/antithesis/configs/base/mzcompose.py
@@ -0,0 +1,119 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+"""
+Source-of-truth composition for the `base` Antithesis scenario.
+
+This is the minimal SQL + Kafka topology: Materialize plus a Kafka-compatible
+broker plus Schema Registry plus a long-lived `antithesis-test-driver` client
+that hosts Test Composer commands and emits `setup_complete` once the system
+is up. Other scenarios (`mysql_mt_replicas`, `pg_cdc`, …) live in sibling
+directories under `antithesis/configs/`; see
+`antithesis/scratchbook/scenario-strategy.md` for when to add one.
+
+The compose file at `antithesis/configs/base/docker-compose.yaml` is the
+artifact `snouty validate`, `snouty launch`, and the Antithesis platform
+consume. Keep it generated and committed; regenerate it with:
+
+    bin/pyactivate antithesis/bin/render-compose-yaml.py --scenario base
+
+(That helper invokes `bin/mzcompose --find base config` to render the
+canonical compose form, then layers on the Antithesis-required attributes
+mzcompose does not emit by default.)
+"""
+
+from __future__ import annotations
+
+from materialize.mzcompose.composition import Composition
+from materialize.mzcompose.service import Service, ServiceConfig
+from materialize.mzcompose.services.materialized import MaterializeEmulator
+from materialize.mzcompose.services.redpanda import Redpanda
+
+# ---------------------------------------------------------------------------
+# Service definitions
+# ---------------------------------------------------------------------------
+
+
+# Use Redpanda as a single-process Kafka + Schema Registry. It exposes both via
+# the `kafka` and `schema-registry` network aliases, which is what the existing
+# upsert_sources workload helper expects.
+REDPANDA = Redpanda(auto_create_topics=True)
+
+# `MaterializeEmulator` is the smallest viable Materialize service for the
+# Antithesis bring-up: it boots with sane defaults, file-backed persist, and
+# the embedded metadata store. Topology-level variations (external metadata,
+# Minio-backed persist, multi-`materialized`) belong in a separate launch.
+MATERIALIZED = MaterializeEmulator()
+
+
+class AntithesisTestDriver(Service):
+    """Long-lived client container that hosts the Antithesis Test Composer
+    commands under `/opt/antithesis/test/v1/`.
+
+    Pulls the workload tree and Materialize Python helpers in via the
+    `antithesis-test-driver` mzbuild image. Container/host names are pinned so
+    Antithesis can attribute logs and faults consistently.
+    """
+
+    def __init__(self, name: str = "antithesis-test-driver") -> None:
+        config: ServiceConfig = {
+            "mzbuild": "antithesis-test-driver",
+            "container_name": name,
+            "hostname": name,
+            "platform": "linux/amd64",
+            "environment": [
+                "NO_COLOR=1",
+                "FORCE_COLOR=0",
+                "PY_COLORS=0",
+                "TERM=dumb",
+                "MZ_ROOT=/opt/materialize",
+                "PYTHONPATH=/opt/materialize/misc/python",
+                "MZ_TESTDRIVE=testdrive",
+                "MZ_MATERIALIZE_URL=postgres://materialize@materialized:6875",
+                "MZ_MATERIALIZE_INTERNAL_URL=postgres://mz_system@materialized:6877",
+                "MZ_KAFKA_ADDR=kafka:9092",
+                "MZ_SCHEMA_REGISTRY_URL=http://schema-registry:8081",
+                "MZ_TESTDRIVE_DEFAULT_TIMEOUT=60s",
+                "MZ_HOST=materialized",
+                "MZ_SQL_PORT=6875",
+                "MZ_HTTP_PORT=6878",
+                "KAFKA_HOST=kafka",
+                "KAFKA_PORT=9092",
+                "SR_HOST=schema-registry",
+                "SR_PORT=8081",
+            ],
+            "depends_on": {
+                "materialized": {"condition": "service_started"},
+                "redpanda": {"condition": "service_started"},
+            },
+        }
+        super().__init__(name=name, config=config)
+
+
+SERVICES = [
+    MATERIALIZED,
+    REDPANDA,
+    AntithesisTestDriver(),
+]
+
+
+# ---------------------------------------------------------------------------
+# Workflows
+# ---------------------------------------------------------------------------
+
+
+def workflow_default(c: Composition) -> None:
+    """Bring the harness up locally and emit `setup_complete`.
+
+    Intended for quick smoke-checks; `snouty validate` performs the same kind
+    of bring-up under its own timeout budget.
+    """
+
+    c.up("materialized", "redpanda", "antithesis-test-driver")
+    c.sql("SELECT 1")
diff --git a/antithesis/scratchbook/deployment-topology.md b/antithesis/scratchbook/deployment-topology.md
new file mode 100644
index 0000000000000..892a22a3e8fac
--- /dev/null
+++ b/antithesis/scratchbook/deployment-topology.md
@@ -0,0 +1,112 @@
+---
+commit: e7ac38b338
+updated: 2026-05-06
+scope: existing-test-driver-integration
+---
+
+# Deployment Topology
+
+This topology is for integrating Materialize's existing test drivers with the
+Antithesis Test Composer. Start small to prove the harness, then move to a
+broad integration topology for source/sink/config coverage. Every additional
+container adds state space, so dependency containers should be introduced in
+deliberate tiers.
+
+## Minimal useful topology
+
+| Container | Role | Image source | Runs | Notes |
+| --- | --- | --- | --- | --- |
+| `materialized` | SUT | Adapt `src/materialized/ci/Dockerfile` | `entrypoint.sh` / `materialized` | Existing image already packages the binary and exposes `environmentd`/`clusterd` symlinks. |
+| `test-driver` | Client | New Antithesis workload image | Long-lived entrypoint plus `/opt/antithesis/test/v1/...` commands | Emits `setup_complete` after health checks, then sleeps so Test Composer can run commands. |
+
+Start with only SQL/table/view tests. Add dependency containers only when a
+driver command needs them.
+
+## Broad integration topology
+
+For the user's intended coverage across many configs, sources, and sinks, the
+useful second topology is intentionally wider:
+
+| Container | Role | Why it exists |
+| --- | --- | --- |
+| `materialized` | SUT | Primary database under test. |
+| `test-driver` | Client | Owns Antithesis test templates, lifecycle signal, assertions, and wrappers. |
+| `redpanda` or `kafka` | Dependency | Kafka source/sink coverage. Prefer one broker implementation per run. |
+| `schema-registry` | Dependency | Avro/CSR source and sink coverage. |
+| `postgres` | Dependency | Postgres CDC and reference checks. |
+| `mysql` | Dependency | MySQL CDC. |
+| `sql-server` | Dependency | SQL Server CDC. |
+| `minio` | Dependency | S3-compatible object storage, COPY, sinks, persist blob/durability testing. |
+| `azurite` | Dependency | Azure blob coverage; use in the same run only if it stays cheap, otherwise split. |
+| `cockroach` or `postgres-metadata` | Dependency | External persist consensus/metadata and durability scenarios. |
+
+Keep topology families that require different startup settings as separate
+Antithesis launches. Examples: file-backed persist vs external metadata,
+Minio-backed persist vs Azurite-backed persist, embedded clusterd vs external
+clusterd, single `materialized` vs 0dt/multi-`materialized`.
+
+## Optional dependencies by driver coverage
+
+| Dependency | Needed for | Existing source |
+| --- | --- | --- |
+| Kafka-compatible broker plus schema registry | Kafka sources/sinks and most full `testdrive` corpus coverage | `test/testdrive/mzcompose.py` currently uses Kafka, Redpanda, Zookeeper, and Schema Registry variants. |
+| Postgres, MySQL, SQL Server | CDC and connection tests | Used by `testdrive`, `parallel-workload`, and `zippy` compositions. |
+| Minio or Azurite | Persist external blob store, S3 COPY, backups, sinks | Existing mzcompose services are already used by `testdrive`, `parallel-workload`, and `zippy`. |
+| CockroachDB or Postgres metadata store | External metadata/persist consensus scenarios | Needed for backup/restore, 0dt, and some durability scenarios. |
+| Polaris/Iceberg | Iceberg source/sink coverage | Used by `parallel-workload` and `zippy`, not needed for the first integration. |
+
+## Test directories for broad coverage
+
+In the broad topology, use multiple test directories rather than one giant
+directory:
+
+```text
+/opt/antithesis/test/v1/base_sql/
+/opt/antithesis/test/v1/kafka_sources/
+/opt/antithesis/test/v1/kafka_sinks/
+/opt/antithesis/test/v1/postgres_cdc/
+/opt/antithesis/test/v1/mysql_cdc/
+/opt/antithesis/test/v1/sql_server_cdc/
+/opt/antithesis/test/v1/s3_copy/
+/opt/antithesis/test/v1/persist_durability/
+/opt/antithesis/test/v1/catalog_ddl/
+```
+
+Each directory owns a coherent setup and command set. Shared helper code belongs
+under `helper_` paths.
+
+## Client image responsibilities
+
+- Include the Antithesis Python SDK for workload assertions and lifecycle
+  signaling.
+- Include `testdrive` if running `.td` scripts directly.
+- Include the Materialize Python package and selected driver modules if running
+  Python workloads (`parallel_workload`, `zippy`, SQLancer orchestration, or
+  custom wrappers).
+- Install test templates under `/opt/antithesis/test/v1/<template>/`.
+- Keep helper code under `helper_` paths so Test Composer ignores it.
+- Keep the container running after readiness, for example with `sleep infinity`.
+
+## Readiness
+
+`setup_complete` belongs in the long-lived client entrypoint, not in a
+`first_` command. The entrypoint should:
+
+1. Wait for `materialized` SQL and HTTP ports.
+2. Wait for any selected dependency services.
+3. Run cheap bootstrapping checks, for example `SELECT 1`.
+4. Emit the Antithesis lifecycle ready signal.
+5. Sleep forever.
+
+## Local validation shape
+
+Local checks should mirror Antithesis command execution:
+
+```text
+docker compose -f antithesis/config/docker-compose.yaml up
+docker compose -f antithesis/config/docker-compose.yaml exec test-driver /opt/antithesis/test/v1/<template>/<command>
+```
+
+Do not run `mzcompose` from inside an Antithesis command. `mzcompose` controls
+Docker and host ports; inside Antithesis, Docker Compose is the orchestrator and
+test commands should behave like clients of the already-running topology.
diff --git a/antithesis/scratchbook/existing-assertions.md b/antithesis/scratchbook/existing-assertions.md
new file mode 100644
index 0000000000000..5eb3173417edd
--- /dev/null
+++ b/antithesis/scratchbook/existing-assertions.md
@@ -0,0 +1,64 @@
+---
+commit: e7ac38b338
+updated: 2026-05-06
+scope: antithesis-sdk-search
+---
+
+# Existing Antithesis Assertions
+
+Before the upsert-source prototype, no existing Antithesis SDK assertions or
+lifecycle calls were found in this worktree.
+
+Search performed:
+
+```text
+rg -n "antithesis_sdk|antithesis\.assertions|antithesis\.lifecycle|assert_always!|assert_sometimes!|assert_reachable!|assert_unreachable!|ANTITHESIS_STOP_FAULTS|ANTITHESIS_SDK" --glob '!target/**' --glob '!Cargo.lock' .
+```
+
+The upsert-source prototype now adds workload-side SDK calls through
+`misc/python/materialize/antithesis/upsert_sources.py`:
+
+| Location | Assertion type | Message |
+| --- | --- | --- |
+| `upsert_sources.py` | `reachable` | `upsert source command saw tolerated testdrive failure` |
+| `upsert_sources.py` | `reachable` | `upsert source setup completed` |
+| `upsert_sources.py` | `sometimes` | `upsert source write driver completed at least once` |
+| `upsert_sources.py` | `sometimes` | `upsert source stale-safe read completed at least once` |
+| `upsert_sources.py` | `sometimes` | `upsert source health check completed at least once` |
+| `upsert_sources.py` | `reachable` | `upsert source expected rows verified` |
+| `upsert_sources.py` | `always` | `upsert source catches up after quiet period` |
+
+There are still no SUT-side Rust assertions. Add those only where a property
+needs internal visibility that the workload cannot observe.
+
+The shared runner module `materialize.antithesis.td_runner` adds two more
+workload-side reachable buckets used by all Tier 1 .td wrappers:
+
+| Location | Assertion type | Message |
+| --- | --- | --- |
+| `td_runner.py` | `reachable` | `td_runner completed .td file` |
+| `td_runner.py` | `reachable` | `td_runner saw tolerated failure` |
+
+The four `testdrive_*` area templates (sql / kafka / load_generator /
+recovery) each have a single `singleton_driver_random_*` command and
+intentionally **no** `eventually_*` recovery check. We tried per-area
+`eventually_*` checks first (initially trivial `count(*) >= 0` queries,
+then a real CREATE/INSERT/SELECT/DROP round-trip), but the round-trip
+ended up being a generic "MZ still works" probe whose pass/fail signal
+was only weakly correlated with the chosen .td's actual semantics — when
+the singleton picks a random file, you can't write a useful recovery
+property without knowing what state was created. That correlation
+belongs in SUT-side Rust assertions or in scenario-specific
+`eventually_*` commands (like `upsert_sources/eventually_*` which has
+its own expected-state model). Add those when there's something
+concrete to verify.
+
+The Antithesis Python SDK is now available locally too: `antithesis==0.2.0`
+is pinned in `ci/builder/requirements.txt` (the real source of
+`misc/python/requirements.txt`, which is a symlink). Both the test-driver
+Docker image and the local uv venv resolve `from antithesis.assertions
+import reachable` to the same real SDK. Our scaffolding directory
+`antithesis/` at the repo root coexists peacefully with the installed
+package because antithesis 0.2.0 is itself a namespace package — Python's
+namespace-package merge picks up submodules from site-packages while our
+directory contributes nothing conflicting.
diff --git a/antithesis/scratchbook/scenario-strategy.md b/antithesis/scratchbook/scenario-strategy.md
new file mode 100644
index 0000000000000..2593f5bea34bb
--- /dev/null
+++ b/antithesis/scratchbook/scenario-strategy.md
@@ -0,0 +1,101 @@
+---
+commit: HEAD
+updated: 2026-05-06
+scope: how-to-add-additional-antithesis-scenarios
+---
+
+# Scenario Strategy
+
+Decision: Antithesis runs are organized as **N narrow scenarios**, each with its
+own compose file and config directory, rather than one wide union topology.
+
+## Why
+
+The Antithesis docs are explicit:
+
+- A test launch consumes exactly one config image (`antithesis.config_image`)
+  and a list of service images (`antithesis.images`). Multiple launches per
+  commit are normal — see the GitHub Action `test_name` field documented at
+  `https://antithesis.com/docs/using_antithesis/ci/`.
+- Within one config, Antithesis picks **exactly one** test template per
+  execution history (`first_test`). Workloads that share a topology can ship as
+  multiple templates; topologies that conflict belong in separate configs.
+- The "Optimizing" page rewards small focused environments. The "virtual time
+  per wall time" metric punishes oversized topologies — the recommendation is
+  that `vt/wt × n_containers > 1`, which one large compose makes hard.
+- The "Sizing" page treats parallelism as the lever for finding bugs faster,
+  not topology breadth.
+
+## Repo layout
+
+```
+antithesis/
+  configs/
+    base/                          # SQL + Kafka/SR + driver
+      docker-compose.yaml
+    mysql_mt_replicas/             # MySQL CDC w/ replica_parallel_workers > 1
+      docker-compose.yaml
+    pg_cdc/
+      docker-compose.yaml
+    ...
+  test/v1/                         # shared template tree
+    upsert_sources/
+    mysql_mts/
+    pg_cdc/
+  test-driver/                     # one mzbuild image; ships the template tree
+  bin/render-compose-yaml.py       # `--scenario <name>` selects which compose to render
+  mzcompose.py                     # one SERVICES list per scenario, exposed via workflows
+```
+
+The driver image stays shared across scenarios. Each scenario's compose
+references it by the same image tag; the cost of carrying templates the
+scenario doesn't need is small relative to building and pushing N driver
+variants.
+
+## When to split vs share
+
+- Different *workload* on the same topology → multiple test templates inside
+  the same config (Antithesis selects one per timeline).
+- Different *topology* (extra/different services, different startup args) →
+  separate config directory, separate `snouty launch`.
+- Different *startup parameter on the same service* (e.g. file-backed persist
+  vs Minio-backed persist; `replica_parallel_workers=1` vs `=8`) → separate
+  scenario. Per-timeline config tunables go inside `first_` setup commands.
+
+## Per-launch identity
+
+Use `snouty launch --source <scenario_name>` so each scenario keeps its own
+property history line in reports. The webhook reference notes property history
+is "generated from all previous runs with the same `antithesis.source`
+parameter."
+
+## Implementation evolution
+
+The branch starts with a single scenario at `antithesis/config/` (legacy
+singular). As soon as the second scenario lands, rename to
+`antithesis/configs/base/` and update `render-compose-yaml.py` to take a
+`--scenario` argument. Don't pre-emptively migrate before there's a second
+consumer.
+
+## Examples to plan around
+
+- **base**: materialized + redpanda + driver; covers Tier 1 SQL + Kafka work.
+- **pg_cdc**: + Postgres; reuses `test/pg-cdc/*.td`.
+- **mysql_cdc**: + MySQL primary; reuses `test/mysql-cdc/*.td`.
+- **mysql_mt_replicas (SS-95)**: + MySQL primary + MySQL replica configured
+  with `replica_parallel_workers=N`. Tests the contract enforced at
+  `src/storage/src/source/mysql/replication.rs:508` under fault injection.
+- **persist_durability**: + cockroach (or postgres metadata) + minio; covers
+  external metadata + external blob; later.
+- **s3_copy**: + minio; reuses `test/testdrive/copy-*-s3-minio*.td`.
+
+## Sources
+
+- `https://antithesis.com/docs/test_templates/first_test/`
+- `https://antithesis.com/docs/test_templates/test_best_practices/`
+- `https://antithesis.com/docs/webhook/test_webhook/`
+- `https://antithesis.com/docs/webhook/webhook_reference/`
+- `https://antithesis.com/docs/getting_started/setup/`
+- `https://antithesis.com/docs/best_practices/sizing/`
+- `https://antithesis.com/docs/best_practices/optimizing/`
+- `https://antithesis.com/docs/using_antithesis/ci/`
diff --git a/antithesis/scratchbook/sut-analysis.md b/antithesis/scratchbook/sut-analysis.md
new file mode 100644
index 0000000000000..f036059654c8c
--- /dev/null
+++ b/antithesis/scratchbook/sut-analysis.md
@@ -0,0 +1,106 @@
+---
+commit: e7ac38b338
+updated: 2026-05-06
+scope: existing-test-driver-integration
+---
+
+# SUT Analysis
+
+This is a targeted analysis for integrating existing Materialize test drivers
+with Antithesis. It is not a full product-wide property catalog.
+
+## System shape
+
+Materialize is exercised in existing system tests as a networked database
+process (`materialized`/`environmentd`) plus optional dependencies such as
+Kafka, Schema Registry, Postgres, MySQL, SQL Server, Minio/Azurite, CockroachDB,
+and Polaris. The existing `src/materialized/ci/Dockerfile` packages the
+`materialized` binary and symlinks it as `environmentd` and `clusterd`, which
+is a useful base for an Antithesis SUT image.
+
+Existing tests are normally orchestrated by `mzcompose`, which starts services,
+sets ports, applies Materialize system parameters, and then runs one or more
+drivers. In Antithesis, Docker Compose owns orchestration, so `mzcompose` should
+be treated as a source of service topology and command arguments rather than as
+a scheduled test command.
+
+## Existing drivers
+
+- `testdrive`: networked integration DSL for Materialize and external systems.
+  The Rust crate exposes `run_file`, `run_stdin`, and `run_string`; the CI image
+  installs the `testdrive` binary. It is the best first target because it can
+  run as a normal executable client against service hostnames.
+- `sqllogictest`: SQL correctness corpus. It is valuable, but less directly
+  useful for distributed Antithesis faults because the current runner embeds an
+  Mz instance rather than exercising the normal multi-container deployment.
+- `parallel-workload`: randomized concurrent SQL stress workload. It is a good
+  Antithesis fit conceptually, but today its `mzcompose` workflow starts many
+  dependencies and its core `run()` path requires a `Composition` for setup and
+  several scenarios.
+- `zippy`: pseudo-random sequential action generator with expected-state
+  validation. It is strong for correctness, but currently uses `mzcompose` for
+  service lifecycle actions and testdrive invocations.
+- `SQLancer`, `SQLsmith`, and `RQG`: query generators/fuzzers. These are good
+  candidates for direct command wrappers after a base client image exists.
+
+## State under test
+
+Relevant durable and semi-durable state includes:
+
+- SQL catalog objects: databases, schemas, roles, clusters, replicas, tables,
+  views, materialized views, sources, sinks, secrets, and connections.
+- Persist state: blob storage and consensus metadata, optionally backed by file
+  storage, Minio/Azurite, CockroachDB, Postgres metadata, or FoundationDB.
+- External data systems: Kafka topics, schema registry subjects, upstream CDC
+  tables, object storage contents, and sink outputs.
+- Workload-local expected state: `parallel-workload`'s shared `Database` object
+  and `zippy`'s generated capabilities/state.
+
+## Concurrency model
+
+Materialize itself has multiple asynchronous subsystems: SQL/session handling,
+catalog/coordinator work, storage ingestion, compute dataflows, persist reads
+and writes, and external source/sink clients. The existing Python
+`parallel-workload` driver adds client-side concurrency by spawning worker
+threads that randomly choose action lists and share a thread-safe model of
+created objects. Zippy instead generates a single sequential action stream and
+validates expected state.
+
+Antithesis can add process, network, timing, and clock faults around these
+drivers. The first integration should avoid duplicating fault systems until the
+base harness is stable.
+
+## Failure-prone integration areas
+
+- Driver commands must eventually exit; several Materialize workloads are
+  designed as long-running CI or longevity runs and need bounded runtimes.
+- Existing workflows often assume `mzcompose` can start, stop, or kill services.
+  Those actions need translation or temporary exclusion under Antithesis.
+- Reset semantics matter. `testdrive` defaults to cleaning Materialize state
+  before scripts, while Antithesis driver commands may intentionally share state
+  within a timeline.
+- External dependencies should be introduced incrementally. Starting with the
+  full `testdrive` or `parallel-workload` service set would add avoidable state
+  space and make early failures harder to classify.
+- Expected transient failures under active fault injection need to be retried or
+  classified. Hard failures should represent property violations or failure to
+  recover after faults are quiet.
+- The desired coverage spans many configurations, sources, and sinks. A single
+  minimal topology will not be enough. The harness needs a tiered strategy:
+  randomize cheap per-timeline configuration inside Test Composer commands, and
+  split startup/topology variations into separate Antithesis launches.
+
+## Integration conclusion
+
+The least risky path is:
+
+1. Package a small client container and run selected networked `testdrive`
+   scripts as singleton or serial driver commands.
+2. Add a broad integration topology with common source/sink dependencies and
+   separate test directories for base SQL, Kafka, CDC, object storage, persist,
+   and catalog/DDL families.
+3. Add direct wrappers for query fuzzers and composition-free
+   `parallel-workload` scenarios.
+4. Decompose the strongest randomized workloads into first, parallel, serial,
+   anytime, eventually, and finally commands after the base harness has proven
+   service discovery, readiness, logging, and recovery checks.
diff --git a/antithesis/scratchbook/test-driver-integration.md b/antithesis/scratchbook/test-driver-integration.md
new file mode 100644
index 0000000000000..64b2f90ea8b3e
--- /dev/null
+++ b/antithesis/scratchbook/test-driver-integration.md
@@ -0,0 +1,301 @@
+---
+commit: e7ac38b338
+updated: 2026-05-06
+scope: existing-test-drivers-to-antithesis-test-composer
+status: research
+---
+
+# Test Driver Integration
+
+## Antithesis driver model
+
+Antithesis Test Composer discovers executable commands under
+`/opt/antithesis/test/v1/<test_dir>/<prefix>_<command>`. The filename prefix
+controls scheduling:
+
+- `first_`: one-time per-timeline setup, no faults, no concurrency.
+- `singleton_driver_`: run one existing monolithic test or workload once per
+  timeline under fault injection.
+- `serial_driver_`: repeat operations that need exclusive driver access.
+- `parallel_driver_`: repeat operations that can overlap with other drivers.
+- `anytime_`: invariant or availability checks that may run alongside drivers.
+- `eventually_`: terminal recovery check after faults and drivers stop.
+- `finally_`: terminal final-state check after drivers complete naturally.
+
+The official porting path for existing tests is: start with a
+`singleton_driver_` wrapper, then split the workload into `first_`,
+`parallel_driver_`, `serial_driver_`, `anytime_`, `eventually_`, and
+`finally_` commands as the workload matures.
+
+Sources:
+
+- https://antithesis.com/docs/test_templates/first_test/
+- https://antithesis.com/docs/test_templates/test_composer_reference/
+- https://antithesis.com/docs/test_templates/composer_example/
+- https://antithesis.com/docs/environment/fault_injection/
+
+## Materialize driver fit
+
+| Driver | Current shape | Antithesis fit | Integration work |
+| --- | --- | --- | --- |
+| `testdrive` | Rust binary plus `.td` corpus. `src/testdrive/src/lib.rs` exposes `run_file`, `run_stdin`, and `run_string`; `src/testdrive/ci/Dockerfile` already packages the binary. | Best first target. Use as `singleton_driver_` or `serial_driver_` around selected `.td` files. | Build a client image that includes `testdrive` and selected `.td` files. Pass service hostnames directly instead of relying on `mzcompose`. Disable reset for commands that share state. |
+| `parallel-workload` | Python randomized concurrent SQL workload. `test/parallel-workload/mzcompose.py` starts many services and calls `materialize.parallel_workload.parallel_workload.run`. | Best existing stress workload for Antithesis once decoupled from `mzcompose`. Initially use `singleton_driver_parallel_workload_regression` with a short runtime. Later split action groups into `parallel_driver_` commands. | Refactor the standalone path: `run()` currently requires a `Composition` during database setup and several scenarios require `composition` for kill, 0dt, backup/restore, Minio, and Polaris actions. Start with `regression`, `rename`, or `repeat-row` scenarios and let Antithesis provide node/network faults. |
+| `zippy` | Sequential randomized, model-like workload that tracks expected state and validates correctness. | Strong correctness driver as a `singleton_driver_`; longer-term, map bootstrap to `first_`, actions to `serial_driver_`, and validation to `anytime_`/`finally_`. | It is tightly coupled to `mzcompose` for service start/stop, testdrive calls, and failure actions. Needs an Antithesis client adapter with topology already running and service-control actions either removed or translated. |
+| `SQLancer` | External generator with Materialize wrapper in `misc/python/materialize/sqlancer.py`. | Good `singleton_driver_` or `parallel_driver_` for SQL logic bugs. | Existing wrapper starts `materialized` through `mzcompose`; create a direct command wrapper targeting the already-running `materialized` service. |
+| `SQLsmith` | Query generator against one or more Materialize instances. | Good `parallel_driver_` query-fuzzing command after basic setup. | Existing mzcompose workflow seeds data and runs a service; turn the seed-data step into `first_` and the generator invocation into `parallel_driver_`. |
+| `RQG` | Grammar-based query generator with optional reference comparison. | Good later-stage `singleton_driver_` or `parallel_driver_`, especially for reference-comparison workloads. | Needs a direct wrapper and a clear decision about whether the reference DB belongs in the topology. |
+| `sqllogictest` | Embedded-Mz SQL correctness runner. | Lower initial value for Antithesis distributed fault testing because it does not exercise the normal multi-container system path. | Can be used for SQL-engine smoke coverage, but prioritize networked drivers first. |
+| `mzcompose` | Python orchestration over Docker Compose. | Use as source material for topology and command arguments, not as an Antithesis test command. | Do not invoke from Test Composer commands, because commands run inside containers while Antithesis owns orchestration. |
+
+## Recommended integration sequence
+
+### Phase 1: thin singleton wrappers
+
+Goal: get existing coverage running under Antithesis faults with minimal
+refactoring.
+
+Create one template, for example `/opt/antithesis/test/v1/materialize_smoke/`,
+in the `test-driver` container:
+
+```text
+first_bootstrap_sql
+singleton_driver_testdrive_smoke
+singleton_driver_sqlancer_norec
+eventually_materialized_recovers
+```
+
+Start with a small `.td` selection that only needs `materialized`, then add
+Kafka/Schema Registry/Minio/Postgres dependencies as soon as the base wiring is
+stable. This proves image layout, service discovery, readiness signaling,
+nonzero-exit failure reporting, and local dry-run mechanics.
+
+### Phase 1b: broad topology, narrow commands
+
+Goal: cover many Materialize configurations, sources, and sinks without turning
+the first Antithesis run into an unstructured full-CI clone.
+
+Use one "broad integration" topology that starts the common dependency set:
+
+```text
+materialized
+test-driver
+redpanda-or-kafka
+schema-registry
+postgres
+mysql
+sql-server
+minio
+azurite
+cockroach-or-postgres-metadata
+```
+
+Within that topology, split incompatible workload families into separate test
+directories. Antithesis runs commands from one test directory per execution
+history, so this gives each history a coherent setup while the overall run
+covers many families.
+
+Suggested directories:
+
+```text
+/opt/antithesis/test/v1/base_sql/
+/opt/antithesis/test/v1/kafka_sources/
+/opt/antithesis/test/v1/kafka_sinks/
+/opt/antithesis/test/v1/postgres_cdc/
+/opt/antithesis/test/v1/mysql_cdc/
+/opt/antithesis/test/v1/sql_server_cdc/
+/opt/antithesis/test/v1/s3_copy/
+/opt/antithesis/test/v1/persist_durability/
+/opt/antithesis/test/v1/catalog_ddl/
+```
+
+Each directory should have its own `first_` setup command that creates only the
+objects that its driver commands need. This avoids having every timeline create
+every source and sink type before useful faulted execution begins.
+
+Do not make each existing `.td` file its own test directory. That would preserve
+the old example-based shape and waste Antithesis' scheduling. Prefer command
+families such as `parallel_driver_ingest`, `parallel_driver_verify`,
+`serial_driver_alter_source`, `anytime_health`, and `eventually_catch_up`.
+
+### Configuration model
+
+Treat configuration variation as three different kinds:
+
+| Variation type | Examples | Where it belongs |
+| --- | --- | --- |
+| Per-command/session | session variables, transaction isolation, target cluster, SQL API vs pgwire, retry/timeout tuning | Randomize inside driver commands. |
+| Per-timeline SQL setup | cluster size/replica count, feature flags via `ALTER SYSTEM`, source/sink options, envelope/format choices, indexes, materialized vs non-materialized views | Choose in `first_` setup commands, record it in a config table, then have drivers read it. |
+| Per-run topology/startup | persist backend, metadata store, embedded vs external clusterd, Kafka vs Redpanda, Minio vs Azurite, single vs multi-`materialized`, boot flags/environment variables | Use separate compose/config images or separate Antithesis launches. |
+
+This prevents one run from needing mutually incompatible startup settings while
+still letting Antithesis explore a large state space inside each run.
+
+For the source/sink matrix, the first broad run should prefer representative
+combinations over exhaustive Cartesian product coverage:
+
+| Axis | Initial representative values |
+| --- | --- |
+| Kafka source envelope | append-only, upsert, Debezium-style where cheap |
+| Kafka format | Avro/CSR and JSON first; add Protobuf later if needed |
+| Kafka sink | exactly-once sink plus simple sink smoke |
+| CDC source | one Postgres, one MySQL, one SQL Server setup path |
+| Object storage | S3-compatible Minio first; Azurite in a separate run or separate directory if both stay cheap |
+| Persist/metadata | file/local for smoke; external metadata/blob for durability runs |
+| Cluster topology | one multi-replica cluster in broad run; separate run for external clusterd/topology-specific faults |
+
+### Phase 2: dedicated randomized workload wrappers
+
+Goal: reuse Materialize's existing randomized drivers while avoiding
+orchestration conflicts.
+
+- Add a composition-free `parallel-workload` command for non-orchestration
+  scenarios. It should target service hostnames, take bounded `--runtime` and
+  `--threads`, and use workload-side Antithesis assertions for query failures,
+  unexpected panics, and meaningful reachability.
+- Add a `zippy` singleton command for a narrow scenario that does not start or
+  stop containers itself.
+- Add `eventually_` or guarded `ANTITHESIS_STOP_FAULTS` recovery checks that
+  poll SQL/HTTP availability and then validate durable state.
+
+### Phase 3: full Test Composer decomposition
+
+Goal: let Antithesis choose operation ordering and concurrency instead of
+running a monolith.
+
+Suggested mapping:
+
+| Test Composer command | Materialize behavior |
+| --- | --- |
+| `first_schema` | Create baseline databases, roles, clusters, connections, seed tables, and optional source topics. |
+| `parallel_driver_insert` | Insert/COPY/Kafka ingest actions from `parallel-workload` and `zippy`. |
+| `parallel_driver_select` | Peeks, subscribes, HTTP SQL, prepared statements, and query fuzzing. |
+| `parallel_driver_ddl` | DDL that is safe to run concurrently with expected-error handling. |
+| `serial_driver_admin` | Mutations that need exclusivity: disruptive DDL, large source/sink setup, backup/restore-like phases. |
+| `anytime_availability` | Low-cost SQL/HTTP health and "read eventually succeeds" checks under active faults. |
+| `eventually_recovery` | Stop faults, poll for recovery, verify sources/sinks/dataflows catch up. |
+| `finally_consistency` | Validate final row counts, watermarks, sink contents, and catalog consistency when drivers completed naturally. |
+
+## Key engineering constraints
+
+- Test commands must eventually exit. Long-running drivers need explicit small
+  runtimes.
+- Commands should tolerate transient network and process failures during driver
+  phases. Reserve hard failures for property violations, unrecoverable errors,
+  or failure to recover after a quiet period.
+- Do not duplicate fault systems at first. Antithesis already injects network,
+  throttling, hang, restart, and clock faults. Materialize actions like kill,
+  0dt, backup/restore, and toxiproxy are useful later, but they should be
+  reintroduced intentionally.
+- Use Antithesis SDK randomness for new wrappers. Existing `random.seed(...)`
+  paths are reproducible locally, but Antithesis-guided randomness gives better
+  replay and search.
+- Prefer workload-side Python assertions first. Add Rust SDK assertions in
+  `materialized` only where internal state is necessary to express the
+  property.
+- Avoid duplicating Materialize test infrastructure. Antithesis commands should
+  be thin adapters over shared helper modules, existing driver entry points, and
+  reusable source/sink setup code. If a wrapper starts accumulating generic
+  Kafka, CDC, SQL retry, expected-error, or object-model logic, move that logic
+  into a reusable `misc/python/materialize/...` helper that both mzcompose tests
+  and Antithesis commands can call.
+
+## Thin-adapter design
+
+The intended layering is:
+
+```text
+Antithesis command file
+  -> antithesis-specific adapter: lifecycle/assertions/random/quiet-periods
+    -> shared Materialize workload helper
+      -> existing driver/library/client code
+        -> Materialize and dependencies
+```
+
+For example, an upsert-source command should not reimplement all testdrive Kafka
+or SQL behavior. It should either:
+
+- invoke `testdrive` with generated or templated `.td` fragments for setup and
+  straightforward checks; or
+- call a shared Python helper that owns topic creation, message production,
+  expected-state tracking, and SQL polling.
+
+The Antithesis-specific layer should stay small:
+
+- choose randomized parameters using Antithesis randomness;
+- call the shared helper;
+- translate helper results into Antithesis assertions;
+- request quiet periods or run recovery checks when appropriate;
+- exit cleanly.
+
+This gives Antithesis a native driver shape without creating a second
+Materialize test framework.
+
+## Reuse plan for existing infrastructure
+
+| Existing infrastructure | Reuse strategy |
+| --- | --- |
+| `testdrive` binary and `.td` DSL | Use directly for scripted setup and checks. Generate small `.td` fragments where needed instead of duplicating SQL/Kafka command behavior in Python. |
+| `Composition.testdrive()` / `run_testdrive_files()` argument knowledge | Extract the command-line argument construction into a reusable helper if Antithesis and mzcompose need the same defaults. Do not call Docker orchestration from Antithesis commands. |
+| `parallel-workload` action/object model | Refactor only the orchestration boundary so actions can run against an already-running topology. Keep action definitions, expected errors, and object tracking in the existing modules. |
+| `zippy` action/state model | Reuse scenario/action/state code, but introduce an execution adapter for "services already exist" rather than forking zippy logic. |
+| Source/sink setup patterns in `.td` files | Promote repeated patterns into templates or helper functions used by both current tests and Antithesis commands. |
+| SQL retry/polling logic | Centralize in shared helper code, especially for eventual catch-up checks. |
+
+The first implementation should probably add an `antithesis`-aware adapter
+module under `misc/python/materialize/` rather than building substantial logic
+under `antithesis/test/`. The files under `antithesis/test/` should mostly be
+small executable entry points and `helper_` imports/templates.
+
+## Implemented prototype: upsert sources
+
+The first prototype follows that thin-adapter design:
+
+```text
+antithesis/test/v1/upsert_sources/*
+  -> materialize.antithesis.upsert_sources
+    -> testdrive binary
+      -> Kafka + Materialize
+```
+
+Files:
+
+- `misc/python/materialize/antithesis/sdk.py`: small wrapper around the
+  Antithesis Python SDK with local fallbacks for import-time and local smoke
+  testing.
+- `misc/python/materialize/antithesis/upsert_sources.py`: reusable upsert-source
+  workload helper. It generates small testdrive fragments for setup, writes,
+  reads, health checks, and final consistency checks.
+- `antithesis/test/v1/upsert_sources/first_create_upsert_source`: creates a
+  text/text Kafka upsert source and an expected-state table.
+- `antithesis/test/v1/upsert_sources/parallel_driver_write_upserts`: generates
+  random keyed upsert/tombstone records and records the latest expected value
+  for keys whose Kafka write command completed.
+- `antithesis/test/v1/upsert_sources/parallel_driver_read_stale_safe`: runs
+  low-cost reads that are safe during active faults.
+- `antithesis/test/v1/upsert_sources/anytime_upsert_source_health`: checks
+  source status without requiring exact catch-up under active faults.
+- `antithesis/test/v1/upsert_sources/eventually_upsert_source_catches_up`:
+  requests a quiet period, writes a sentinel, waits for it, and verifies all
+  expected rows are visible.
+- `antithesis/test/v1/upsert_sources/finally_upsert_expected_rows_visible`:
+  verifies the expected-state table is represented in the upsert source table
+  when drivers complete naturally.
+
+This is intentionally not a full testdrive replacement. It delegates Kafka
+topic creation, Kafka ingest, SQL execution, retries, and SQL result checking to
+the existing `testdrive` binary.
+
+## Immediate next implementation slice
+
+1. Create `antithesis/config/docker-compose.yaml` with `materialized` and a
+   `test-driver` client.
+2. Build a `test-driver` image containing the Antithesis Python SDK,
+   `testdrive`, selected `.td` files, and small Python wrappers.
+3. Add a readiness entrypoint that waits for Materialize, emits
+   `setup_complete`, and sleeps.
+4. Add `singleton_driver_testdrive_smoke` for one or two stable SQL-only
+   `.td` files.
+5. Add `eventually_materialized_recovers` that requests or relies on a quiet
+   period, polls Materialize, and checks a small durable invariant.
+6. Validate locally with `docker compose exec` and then with `snouty validate`
+   once `snouty` is installed.
diff --git a/antithesis/setup-complete.sh b/antithesis/setup-complete.sh
new file mode 100755
index 0000000000000..105a2c5423442
--- /dev/null
+++ b/antithesis/setup-complete.sh
@@ -0,0 +1,32 @@
+#!/usr/bin/env bash
+
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Reference fallback that emits Antithesis's `setup_complete` lifecycle event
+# to the SDK output file. The test-driver entrypoint emits the same event
+# through `antithesis.lifecycle.setup_complete` (the canonical path); this
+# script exists for any container that cannot link the Python SDK directly.
+#
+# Antithesis sets `ANTITHESIS_OUTPUT_DIR` automatically. Outside Antithesis,
+# `ANTITHESIS_SDK_LOCAL_OUTPUT` may override the destination.
+
+set -euo pipefail
+
+OUTPUT_PATH="/tmp/antithesis_sdk.jsonl"
+if [[ -n "${ANTITHESIS_OUTPUT_DIR:-}" ]]; then
+  OUTPUT_PATH="${ANTITHESIS_OUTPUT_DIR}/sdk.jsonl"
+  echo "Running in Antithesis, emitting setup_complete to ${OUTPUT_PATH}"
+elif [[ -n "${ANTITHESIS_SDK_LOCAL_OUTPUT:-}" ]]; then
+  OUTPUT_PATH="${ANTITHESIS_SDK_LOCAL_OUTPUT}"
+  echo "Antithesis SDK local output override detected, emitting setup_complete to ${OUTPUT_PATH}"
+fi
+
+mkdir -p "$(dirname "$OUTPUT_PATH")"
+echo '{"antithesis_setup":{"status":"complete","details":{"message":"ready to go"}}}' >> "${OUTPUT_PATH}"
diff --git a/antithesis/test-driver/.gitignore b/antithesis/test-driver/.gitignore
new file mode 100644
index 0000000000000..a2a0a53e2783c
--- /dev/null
+++ b/antithesis/test-driver/.gitignore
@@ -0,0 +1,5 @@
+# Files copied in by `pre-image: copy` plugins. Excluded from the mzbuild
+# fingerprint and removed before each build.
+/misc
+/antithesis
+/td
diff --git a/antithesis/test-driver/Dockerfile b/antithesis/test-driver/Dockerfile
new file mode 100644
index 0000000000000..8357e45b18e14
--- /dev/null
+++ b/antithesis/test-driver/Dockerfile
@@ -0,0 +1,91 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Antithesis Test Composer client container.
+#
+# Inherits the testdrive image so the `testdrive` binary, postgres client,
+# protobuf, and libduckdb are already in place. Adds Python plus the Antithesis
+# Python SDK, copies the workload helpers and test commands, and installs a
+# readiness entrypoint that emits `setup_complete` once the SUT is up.
+
+MZFROM testdrive
+
+USER root
+
+# Disable color output everywhere (Antithesis stores raw bytes; ANSI escapes
+# are garbage in logs and triage).
+ENV NO_COLOR=1
+ENV FORCE_COLOR=0
+ENV PY_COLORS=0
+ENV CLICOLOR=0
+ENV TERM=dumb
+
+RUN apt-get update \
+ && TZ=UTC DEBIAN_FRONTEND=noninteractive apt-get -y install --no-install-recommends \
+        python3 python3-pip ca-certificates curl netcat-openbsd \
+ && apt-get clean \
+ && rm -rf /var/lib/apt/lists/*
+
+# Install the Antithesis Python SDK.
+# Pinned to the latest stable release at image-creation time. Bump in lockstep
+# with `misc/python/materialize/antithesis/sdk.py` import expectations.
+RUN pip3 install --no-cache-dir --break-system-packages antithesis==0.2.0
+
+# Layout convention: MZ_ROOT mirrors the repo root, so existing helpers that
+# look up `MZ_ROOT/misc/python` continue to work.
+COPY misc/python /opt/materialize/misc/python
+COPY antithesis/test /opt/materialize/antithesis/test
+
+# Subset of test/testdrive/*.td files referenced by the Tier 1 wrappers. The
+# mzbuild.yml `pre-image: copy` blocks restrict which files are pulled in, so
+# only those land here; full test/testdrive/ is intentionally not bundled.
+COPY td/testdrive /opt/materialize/td/testdrive
+
+# Antithesis Test Composer command discovery.
+RUN mkdir -p /opt/antithesis/test \
+ && cp -R /opt/materialize/antithesis/test/v1 /opt/antithesis/test/v1 \
+ && chmod -R +x /opt/antithesis/test/v1 \
+ && find /opt/antithesis/test/v1 -name 'helper_*' -prune -o -type f -name '*.py' -print -exec chmod -x {} +
+
+# Antithesis Python SDK assertion cataloging. Symlinking under
+# /opt/antithesis/catalog/ lets the platform discover .py files that contain
+# SDK calls without duplicating them on disk.
+RUN mkdir -p /opt/antithesis/catalog \
+ && ln -s /opt/materialize/misc/python/materialize/antithesis /opt/antithesis/catalog/materialize_antithesis \
+ && ln -s /opt/antithesis/test/v1 /opt/antithesis/catalog/test_v1
+
+ENV MZ_ROOT=/opt/materialize
+# Two roots on PYTHONPATH: misc/python so `materialize.antithesis.*` resolves,
+# and the test/v1 directory so command files can `from helper_bootstrap
+# import …` regardless of which template directory they live in.
+ENV PYTHONPATH=/opt/antithesis/test/v1:/opt/materialize/misc/python
+ENV PATH=/opt/antithesis/test/v1:$PATH
+
+# Default workload-side environment so the upsert_sources commands can find
+# Materialize, Kafka, and the Schema Registry. docker-compose can override.
+ENV MZ_TESTDRIVE=testdrive
+ENV MZ_MATERIALIZE_URL=postgres://materialize@materialized:6875
+ENV MZ_MATERIALIZE_INTERNAL_URL=postgres://mz_system@materialized:6877
+ENV MZ_KAFKA_ADDR=kafka:9092
+ENV MZ_SCHEMA_REGISTRY_URL=http://schema-registry:8081
+ENV MZ_TESTDRIVE_DEFAULT_TIMEOUT=60s
+ENV MZ_HOST=materialized
+ENV MZ_SQL_PORT=6875
+ENV MZ_HTTP_PORT=6878
+ENV KAFKA_HOST=kafka
+ENV KAFKA_PORT=9092
+ENV SR_HOST=schema-registry
+ENV SR_PORT=8081
+
+COPY entrypoint.sh /usr/local/bin/antithesis-test-driver-entrypoint
+RUN chmod +x /usr/local/bin/antithesis-test-driver-entrypoint
+
+WORKDIR /opt/antithesis
+
+ENTRYPOINT ["/usr/local/bin/antithesis-test-driver-entrypoint"]
diff --git a/antithesis/test-driver/entrypoint.sh b/antithesis/test-driver/entrypoint.sh
new file mode 100755
index 0000000000000..c0753c269ad49
--- /dev/null
+++ b/antithesis/test-driver/entrypoint.sh
@@ -0,0 +1,105 @@
+#!/usr/bin/env bash
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Long-lived entrypoint for the Antithesis Test Composer client.
+#
+# Responsibilities:
+#   1. Wait for materialized SQL/HTTP and the Kafka + Schema Registry endpoints.
+#   2. Emit a bootstrap reachable assertion through the Antithesis Python SDK.
+#   3. Emit the lifecycle `setup_complete` signal.
+#   4. Sleep so Antithesis Test Composer can exec individual commands against
+#      this container.
+
+set -euo pipefail
+
+export NO_COLOR=1
+export FORCE_COLOR=0
+export PY_COLORS=0
+
+MZ_HOST="${MZ_HOST:-materialized}"
+MZ_SQL_PORT="${MZ_SQL_PORT:-6875}"
+MZ_HTTP_PORT="${MZ_HTTP_PORT:-6878}"
+KAFKA_HOST="${KAFKA_HOST:-kafka}"
+KAFKA_PORT="${KAFKA_PORT:-9092}"
+SR_HOST="${SR_HOST:-schema-registry}"
+SR_PORT="${SR_PORT:-8081}"
+
+# Total budget for the entire readiness phase. Antithesis bring-up under fault
+# injection can be slow; bias toward a generous deadline.
+WAIT_DEADLINE_SECONDS="${MZ_ANTITHESIS_WAIT_DEADLINE:-600}"
+
+start_epoch=$(date +%s)
+deadline=$(( start_epoch + WAIT_DEADLINE_SECONDS ))
+
+now() { date +%s; }
+
+wait_tcp() {
+    local host=$1 port=$2 desc=$3
+    while ! nc -z "$host" "$port" 2>/dev/null; do
+        if [[ "$(now)" -gt "$deadline" ]]; then
+            echo "[entrypoint] TIMEOUT waiting for $desc ($host:$port)" >&2
+            exit 1
+        fi
+        sleep 1
+    done
+    echo "[entrypoint] ready: $desc ($host:$port)"
+}
+
+wait_http_ok() {
+    local url=$1 desc=$2
+    while ! curl -fsS -o /dev/null "$url"; do
+        if [[ "$(now)" -gt "$deadline" ]]; then
+            echo "[entrypoint] TIMEOUT waiting for $desc ($url)" >&2
+            exit 1
+        fi
+        sleep 1
+    done
+    echo "[entrypoint] ready: $desc ($url)"
+}
+
+echo "[entrypoint] waiting for SUT and dependencies"
+
+wait_tcp  "$MZ_HOST"    "$MZ_SQL_PORT"   "materialized SQL"
+wait_tcp  "$MZ_HOST"    "$MZ_HTTP_PORT"  "materialized HTTP"
+wait_http_ok "http://${MZ_HOST}:${MZ_HTTP_PORT}/api/readyz" "materialized readyz"
+
+wait_tcp  "$KAFKA_HOST" "$KAFKA_PORT"    "kafka"
+wait_tcp  "$SR_HOST"    "$SR_PORT"       "schema-registry"
+
+# Smoke-check the SQL connection so missing auth/config surfaces here, not in
+# every test command.
+PGPASSWORD='' psql -h "$MZ_HOST" -p "$MZ_SQL_PORT" -U materialize -d materialize \
+    -v ON_ERROR_STOP=1 -c "SELECT 1;" >/dev/null
+echo "[entrypoint] materialized accepts SQL"
+
+# Bootstrap: prove that the Antithesis Python SDK is reachable from inside this
+# container, then emit setup_complete. Both calls go through
+# materialize.antithesis.sdk so they tolerate running outside Antithesis (the
+# wrapper falls back to local stubs).
+python3 - <<'PYEOF'
+import os
+import sys
+
+sys.path.insert(0, os.environ.get("PYTHONPATH", "/opt/materialize/misc/python"))
+
+from materialize.antithesis.sdk import reachable, setup_complete
+
+reachable(
+    "antithesis test-driver entrypoint reached",
+    {"phase": "bootstrap"},
+)
+setup_complete({"service": "antithesis-test-driver"})
+PYEOF
+
+echo "[entrypoint] setup_complete emitted; sleeping"
+
+# Stay alive so Antithesis Test Composer can exec commands. `sleep infinity`
+# is a foreground process so signals propagate cleanly.
+exec sleep infinity
diff --git a/antithesis/test-driver/mzbuild.yml b/antithesis/test-driver/mzbuild.yml
new file mode 100644
index 0000000000000..a4e41eacf0256
--- /dev/null
+++ b/antithesis/test-driver/mzbuild.yml
@@ -0,0 +1,41 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Image used as the Antithesis Test Composer client container.
+#
+# It bundles:
+#   - the `testdrive` binary (via MZFROM testdrive)
+#   - the Antithesis Python SDK
+#   - misc/python (so `materialize.antithesis.*` and shared workload helpers
+#     are importable)
+#   - antithesis/test/v1 (so commands are discoverable at
+#     /opt/antithesis/test/v1)
+#   - the entire `test/testdrive/*.td` corpus, so the area templates under
+#     `antithesis/test/v1/testdrive_*/` can pick from
+#     `materialize.antithesis.testdrive_corpus.BASE_*` lists at runtime.
+#
+# Cost note: any change to any .td file in `test/testdrive/` invalidates
+# this image's mzbuild fingerprint. That's intentional — when the testdrive
+# corpus changes, we want the image to pick up the new content. CI's GHCR
+# cache + Docker's layer cache absorb the rebuild cost.
+name: antithesis-test-driver
+publish: false
+pre-image:
+  - type: copy
+    source: misc/python
+    destination: misc/python
+    matching: "**"
+  - type: copy
+    source: antithesis/test
+    destination: antithesis/test
+    matching: "**"
+  - type: copy
+    source: test/testdrive
+    destination: td/testdrive
+    matching: "*.td"
diff --git a/antithesis/test/v1/helper_bootstrap.py b/antithesis/test/v1/helper_bootstrap.py
new file mode 100755
index 0000000000000..4170f79279b39
--- /dev/null
+++ b/antithesis/test/v1/helper_bootstrap.py
@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+
+def add_materialize_python_to_path() -> None:
+    mz_root = Path(os.environ.get("MZ_ROOT", "/workdir"))
+    sys.path.insert(0, str(mz_root / "misc" / "python"))
diff --git a/antithesis/test/v1/testdrive_kafka/singleton_driver_random_kafka b/antithesis/test/v1/testdrive_kafka/singleton_driver_random_kafka
new file mode 100755
index 0000000000000..0ab4556b6e4a6
--- /dev/null
+++ b/antithesis/test/v1/testdrive_kafka/singleton_driver_random_kafka
@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Antithesis area template — Kafka source/sink + Schema Registry.
+#
+# Picks one .td from `BASE_KAFKA` and runs it. The corpus covers
+# UPSERT/NONE/DEBEZIUM envelopes, Avro/JSON/Protobuf formats, multi-partition
+# sinks, exactly-once sinks, the Avro resolution matrix, progress and remap
+# behavior, and topic-recreate semantics. High fault-injection ROI: broker
+# restarts, partial network partitions, and CSR connectivity all flow through
+# this template.
+
+from helper_bootstrap import add_materialize_python_to_path
+
+add_materialize_python_to_path()
+
+from materialize.antithesis.sdk import random_choice
+from materialize.antithesis.td_runner import run_file
+from materialize.antithesis.testdrive_corpus import BASE_KAFKA
+
+chosen = random_choice(BASE_KAFKA)
+run_file(
+    f"/opt/materialize/td/testdrive/{chosen}",
+    source=f"antithesis/testdrive_kafka/singleton_driver_random_kafka:{chosen}",
+)
diff --git a/antithesis/test/v1/testdrive_load_generator/singleton_driver_random_load_generator b/antithesis/test/v1/testdrive_load_generator/singleton_driver_random_load_generator
new file mode 100755
index 0000000000000..89503bbfce44f
--- /dev/null
+++ b/antithesis/test/v1/testdrive_load_generator/singleton_driver_random_load_generator
@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Antithesis area template — LOAD GENERATOR sources.
+#
+# Picks one .td from `BASE_LOAD_GENERATOR`. These exercise the storage
+# ingest control plane and dataflow path independently of Kafka, so failures
+# under faults isolate to the MZ side rather than involving an external
+# broker.
+
+from helper_bootstrap import add_materialize_python_to_path
+
+add_materialize_python_to_path()
+
+from materialize.antithesis.sdk import random_choice
+from materialize.antithesis.td_runner import run_file
+from materialize.antithesis.testdrive_corpus import BASE_LOAD_GENERATOR
+
+chosen = random_choice(BASE_LOAD_GENERATOR)
+run_file(
+    f"/opt/materialize/td/testdrive/{chosen}",
+    source=(
+        "antithesis/testdrive_load_generator/"
+        f"singleton_driver_random_load_generator:{chosen}"
+    ),
+)
diff --git a/antithesis/test/v1/testdrive_recovery/singleton_driver_random_recovery b/antithesis/test/v1/testdrive_recovery/singleton_driver_random_recovery
new file mode 100755
index 0000000000000..c23ec5b4f9862
--- /dev/null
+++ b/antithesis/test/v1/testdrive_recovery/singleton_driver_random_recovery
@@ -0,0 +1,39 @@
+#!/usr/bin/env python3
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Antithesis area template — files chosen specifically because their
+# property is "the system survives and recovers from a perturbation."
+# Examples: source progress/recovery (`kafka-progress.td`), hydration
+# status (`hydration-status.td`), topic recreate (`kafka-recreate-topic.td`),
+# dataflow GC (`dataflow-cleanup.td`), sink progress (`kafka-sink-empty-
+# progress-topic.td`).
+#
+# This area deliberately overlaps with `testdrive_kafka` and
+# `testdrive_load_generator`: the split is for Antithesis triage attribution,
+# not exclusivity. Landing on a recovery-focused file via this template
+# means triage flags it as recovery work even when the same file might also
+# have been picked from another bucket.
+
+from helper_bootstrap import add_materialize_python_to_path
+
+add_materialize_python_to_path()
+
+from materialize.antithesis.sdk import random_choice
+from materialize.antithesis.td_runner import run_file
+from materialize.antithesis.testdrive_corpus import BASE_RECOVERY
+
+chosen = random_choice(BASE_RECOVERY)
+run_file(
+    f"/opt/materialize/td/testdrive/{chosen}",
+    source=(
+        "antithesis/testdrive_recovery/"
+        f"singleton_driver_random_recovery:{chosen}"
+    ),
+)
diff --git a/antithesis/test/v1/testdrive_sql/singleton_driver_random_sql b/antithesis/test/v1/testdrive_sql/singleton_driver_random_sql
new file mode 100755
index 0000000000000..38399e415c97e
--- /dev/null
+++ b/antithesis/test/v1/testdrive_sql/singleton_driver_random_sql
@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Antithesis area template — pure SQL.
+#
+# Each timeline picks one .td from `BASE_SQL` (curated in
+# `materialize.antithesis.testdrive_corpus`) and runs it via the shared
+# `td_runner`. Selection goes through the Antithesis SDK's `random_choice`
+# so platform replay and search interact with which file we land on.
+#
+# Property: the chosen .td completes successfully under fault injection.
+# Coverage spans tables, DDL, joins, transactions, system catalog reads,
+# EXPLAIN, and runtime-error semantics.
+
+from helper_bootstrap import add_materialize_python_to_path
+
+add_materialize_python_to_path()
+
+from materialize.antithesis.sdk import random_choice
+from materialize.antithesis.td_runner import run_file
+from materialize.antithesis.testdrive_corpus import BASE_SQL
+
+chosen = random_choice(BASE_SQL)
+run_file(
+    f"/opt/materialize/td/testdrive/{chosen}",
+    source=f"antithesis/testdrive_sql/singleton_driver_random_sql:{chosen}",
+)
diff --git a/antithesis/test/v1/upsert_sources/anytime_upsert_source_health b/antithesis/test/v1/upsert_sources/anytime_upsert_source_health
new file mode 100755
index 0000000000000..e93e08e6ddf04
--- /dev/null
+++ b/antithesis/test/v1/upsert_sources/anytime_upsert_source_health
@@ -0,0 +1,17 @@
+#!/usr/bin/env python3
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+from helper_bootstrap import add_materialize_python_to_path
+
+add_materialize_python_to_path()
+
+from materialize.antithesis.upsert_sources import health_check
+
+health_check(allow_failure=True)
diff --git a/antithesis/test/v1/upsert_sources/eventually_upsert_source_catches_up b/antithesis/test/v1/upsert_sources/eventually_upsert_source_catches_up
new file mode 100755
index 0000000000000..a0f7234825c8f
--- /dev/null
+++ b/antithesis/test/v1/upsert_sources/eventually_upsert_source_catches_up
@@ -0,0 +1,17 @@
+#!/usr/bin/env python3
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+from helper_bootstrap import add_materialize_python_to_path
+
+add_materialize_python_to_path()
+
+from materialize.antithesis.upsert_sources import write_sentinel_and_verify
+
+write_sentinel_and_verify()
diff --git a/antithesis/test/v1/upsert_sources/finally_upsert_expected_rows_visible b/antithesis/test/v1/upsert_sources/finally_upsert_expected_rows_visible
new file mode 100755
index 0000000000000..757be382a3ddc
--- /dev/null
+++ b/antithesis/test/v1/upsert_sources/finally_upsert_expected_rows_visible
@@ -0,0 +1,19 @@
+#!/usr/bin/env python3
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+from helper_bootstrap import add_materialize_python_to_path
+
+add_materialize_python_to_path()
+
+from materialize.antithesis.upsert_sources import verify_expected_rows_visible
+
+verify_expected_rows_visible(
+    source="antithesis/upsert_sources/finally_upsert_expected_rows_visible"
+)
diff --git a/antithesis/test/v1/upsert_sources/first_create_upsert_source b/antithesis/test/v1/upsert_sources/first_create_upsert_source
new file mode 100755
index 0000000000000..09ae11ea0e6f5
--- /dev/null
+++ b/antithesis/test/v1/upsert_sources/first_create_upsert_source
@@ -0,0 +1,17 @@
+#!/usr/bin/env python3
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+from helper_bootstrap import add_materialize_python_to_path
+
+add_materialize_python_to_path()
+
+from materialize.antithesis.upsert_sources import setup_text_upsert_source
+
+setup_text_upsert_source()
diff --git a/antithesis/test/v1/upsert_sources/parallel_driver_read_stale_safe b/antithesis/test/v1/upsert_sources/parallel_driver_read_stale_safe
new file mode 100755
index 0000000000000..6cfd4a2d51d24
--- /dev/null
+++ b/antithesis/test/v1/upsert_sources/parallel_driver_read_stale_safe
@@ -0,0 +1,17 @@
+#!/usr/bin/env python3
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+from helper_bootstrap import add_materialize_python_to_path
+
+add_materialize_python_to_path()
+
+from materialize.antithesis.upsert_sources import read_stale_safe
+
+read_stale_safe(allow_failure=True)
diff --git a/antithesis/test/v1/upsert_sources/parallel_driver_write_upserts b/antithesis/test/v1/upsert_sources/parallel_driver_write_upserts
new file mode 100755
index 0000000000000..06ecc26fbe7ec
--- /dev/null
+++ b/antithesis/test/v1/upsert_sources/parallel_driver_write_upserts
@@ -0,0 +1,17 @@
+#!/usr/bin/env python3
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+from helper_bootstrap import add_materialize_python_to_path
+
+add_materialize_python_to_path()
+
+from materialize.antithesis.upsert_sources import write_random_upserts
+
+write_random_upserts(allow_failure=True)
diff --git a/ci/builder/requirements.txt b/ci/builder/requirements.txt
index 9c73a71302649..f0df885387d84 100644
--- a/ci/builder/requirements.txt
+++ b/ci/builder/requirements.txt
@@ -12,6 +12,12 @@
 --only-binary=confluent-kafka
 
 aiohttp==3.13.5
+# Antithesis Python SDK. Used by the workload helpers under
+# materialize.antithesis.* both locally (so type checkers see real types and
+# ad-hoc imports work in the venv) and inside the antithesis-test-driver
+# image (which pins the same version via pip3). Keep this version in sync
+# with antithesis/test-driver/Dockerfile.
+antithesis==0.2.0
 black==26.3.1
 # Pulls in dependencies that make launchdarkly/cloud tests fail, keep for now
 boto3-stubs[ec2,iam,kinesis,s3,sqs,ssm,sts,s3tables]==1.41.5
diff --git a/ci/test/lint-main/checks/check-mzcompose-files.sh b/ci/test/lint-main/checks/check-mzcompose-files.sh
index fc20ea28e42bf..cf9577fe37aa1 100755
--- a/ci/test/lint-main/checks/check-mzcompose-files.sh
+++ b/ci/test/lint-main/checks/check-mzcompose-files.sh
@@ -26,6 +26,7 @@ check_all_files_referenced_in_ci() {
         -not -wholename "./test/console/mzcompose.py" `# Only run manually` \
         -not -wholename "./test/mzcompose_examples/mzcompose.py" `# Example only` \
         -not -wholename "./test/get-cloud-hostname/mzcompose.py" `# Utility, no test` \
+        -not -path "./antithesis/configs/*/mzcompose.py" `# Submitted via snouty to Antithesis, not Buildkite` \
         | sed -e "s|.*/\([^/]*\)/mzcompose.py|\1|")
     while read -r composition; do
         if ! grep -q "composition: $composition" ci/*/pipeline.template.yml; then
diff --git a/misc/completions/bash/_mzcompose b/misc/completions/bash/_mzcompose
index 682fa4e153573..98eb6facaddd6 100644
--- a/misc/completions/bash/_mzcompose
+++ b/misc/completions/bash/_mzcompose
@@ -43,7 +43,7 @@ _shtab_mzcompose_completion_option_strings=('-h' '--help')
 
 _shtab_mzcompose_pos_0_choices=('build' 'config' 'cp' 'create' 'describe' 'ls' 'list' 'description' 'down' 'events' 'exec' 'gen-shortcuts' 'help' 'images' 'kill' 'list-compositions' 'list-workflows' 'logs' 'pause' 'port' 'ps' 'pull' 'push' 'restart' 'rm' 'run' 'scale' 'start' 'stop' 'sql' 'top' 'unpause' 'up' 'web' 'completion')
 _shtab_mzcompose___sanitizer_choices=('address' 'hwaddress' 'cfi' 'thread' 'leak' 'undefined' 'none')
-_shtab_mzcompose___arch_choices=('X86_64' 'AARCH64')
+_shtab_mzcompose___arch_choices=('x86_64' 'aarch64')
 _shtab_mzcompose_completion_pos_0_choices=('bash' 'zsh' 'tcsh')
 
 _shtab_mzcompose_pos_0_nargs=A...
diff --git a/misc/completions/zsh/_mzcompose b/misc/completions/zsh/_mzcompose
index 91141d255eab8..a2f3404a7920e 100644
--- a/misc/completions/zsh/_mzcompose
+++ b/misc/completions/zsh/_mzcompose
@@ -58,7 +58,7 @@ _shtab_mzcompose_options=(
   "--optimized[build Rust binaries with the optimized profile (optimizations, no LTO, no debug symbols)]"
   "--coverage[whether to enable code coverage compilation flags]"
   "--sanitizer[whether to enable a sanitizer]:sanitizer:(address hwaddress cfi thread leak undefined none)"
-  "--arch[the CPU architecture to build for]:arch:(X86_64 AARCH64)"
+  "--arch[the CPU architecture to build for]:arch:(x86_64 aarch64)"
   "--image-registry[the Docker image registry to pull images from and push images to]:image_registry:"
   "--image-prefix[a prefix to apply to all Docker image names]:image_prefix:"
 )
diff --git a/misc/python/materialize/antithesis/__init__.py b/misc/python/materialize/antithesis/__init__.py
new file mode 100644
index 0000000000000..00a2ba7272a3b
--- /dev/null
+++ b/misc/python/materialize/antithesis/__init__.py
@@ -0,0 +1,10 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+"""Helpers for Materialize workloads that run under Antithesis."""
diff --git a/misc/python/materialize/antithesis/sdk.py b/misc/python/materialize/antithesis/sdk.py
new file mode 100644
index 0000000000000..a88ac4915dd05
--- /dev/null
+++ b/misc/python/materialize/antithesis/sdk.py
@@ -0,0 +1,83 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+from __future__ import annotations
+
+import os
+import random
+import subprocess
+from collections.abc import Mapping
+from typing import Any
+
+try:
+    from antithesis.assertions import (  # pyright: ignore[reportMissingTypeStubs]
+        always,
+        reachable,
+        sometimes,
+        unreachable,
+    )
+except Exception:
+
+    def always(condition: bool, message: str, details: Mapping[str, Any]) -> None:
+        if not condition:
+            raise AssertionError(f"{message}: {details}")
+
+    def sometimes(condition: bool, message: str, details: Mapping[str, Any]) -> None:
+        # Local fallback: do not enforce hit aggregation outside Antithesis.
+        _ = condition, message, details
+
+    def reachable(message: str, details: Mapping[str, Any]) -> None:
+        _ = message, details
+
+    def unreachable(message: str, details: Mapping[str, Any]) -> None:
+        raise AssertionError(f"{message}: {details}")
+
+
+try:
+    from antithesis.lifecycle import (  # pyright: ignore[reportMissingTypeStubs]
+        setup_complete,
+    )
+except Exception:
+
+    def setup_complete(details: Mapping[str, Any]) -> None:
+        print(f"setup_complete: {dict(details)}")
+
+
+try:
+    from antithesis.random import (  # pyright: ignore[reportMissingTypeStubs]
+        get_random,
+        random_choice,
+    )
+except Exception:
+
+    def get_random() -> int:
+        return random.getrandbits(64)
+
+    # Match the upstream antithesis.random signature, which is loosely typed
+    # as `(things: list[Any]) -> Any`. Tightening to a generic `Sequence[T]`
+    # here would make pyright complain about the signature drift between the
+    # real SDK and our fallback.
+    def random_choice(things: list[Any]) -> Any:
+        return random.choice(things)
+
+
+def random_int(max_exclusive: int) -> int:
+    assert max_exclusive > 0
+    return get_random() % max_exclusive
+
+
+def stop_faults(duration_seconds: int) -> bool:
+    """Request an Antithesis quiet period, if running inside Antithesis."""
+
+    stop_faults_bin = os.environ.get("ANTITHESIS_STOP_FAULTS")
+    if not stop_faults_bin:
+        return False
+
+    subprocess.run([stop_faults_bin, str(duration_seconds)], check=True)
+    return True
diff --git a/misc/python/materialize/antithesis/td_runner.py b/misc/python/materialize/antithesis/td_runner.py
new file mode 100644
index 0000000000000..ac529990c96a2
--- /dev/null
+++ b/misc/python/materialize/antithesis/td_runner.py
@@ -0,0 +1,109 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+"""Generic `testdrive` runner for Antithesis Test Composer commands.
+
+Most Tier 1 wrappers — anything where the existing `.td` corpus already
+expresses the property we want under fault injection — should be able to
+reduce to a 3-line command file:
+
+    from materialize.antithesis.td_runner import run_file
+
+    run_file("/opt/materialize/td/testdrive/foo.td",
+             source="antithesis/foo/singleton_driver_foo")
+
+This module owns the subprocess invocation, the tolerated-failure path, and
+the SDK assertions that mark which outcome the runner reached.
+
+Why two distinct `reachable` messages: Antithesis aggregates by message, and
+we want the `td_runner saw tolerated failure` bucket to be visible in triage
+even on timelines that ultimately succeed on retry. Keeping the success and
+tolerated-failure messages distinct lets us tell those apart at a glance.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from collections.abc import Mapping
+from pathlib import Path
+
+from materialize.antithesis.sdk import reachable
+from materialize.antithesis.testdrive_config import TestdriveConfig
+
+
+def run_file(
+    td_path: str | Path,
+    *,
+    source: str,
+    config: TestdriveConfig | None = None,
+    vars: Mapping[str, str] | None = None,
+    no_reset: bool = True,
+    tolerate_transient: bool = True,
+    timeout_seconds: float = 600.0,
+) -> None:
+    """Run a single `.td` file via testdrive against the running SUT.
+
+    `tolerate_transient=True` retries once on subprocess failure or timeout,
+    after emitting a `reachable` assertion that records the tolerated failure.
+    A second failure is allowed to propagate as a non-zero exit so Antithesis
+    flags the timeline.
+
+    `source` should match the Test Composer command path (relative under
+    `/opt/antithesis/test/v1/`) so testdrive's `--source` flag attributes
+    output correctly in triage.
+
+    `vars` is forwarded as `--var=key=value` and is the right place for
+    anything you need to override per-call (e.g. small data sizes for
+    bounded-runtime variants).
+    """
+
+    config = config or TestdriveConfig.from_env()
+    td_path = Path(td_path)
+
+    cmd = config.base_command(no_reset=no_reset, source=source)
+    if vars:
+        for key, value in vars.items():
+            cmd.append(f"--var={key}={value}")
+    cmd.append(str(td_path))
+
+    attempts = 2 if tolerate_transient else 1
+    last_error: BaseException | None = None
+
+    for attempt in range(attempts):
+        try:
+            subprocess.run(cmd, check=True, timeout=timeout_seconds)
+            reachable(
+                "td_runner completed .td file",
+                {
+                    "td_path": str(td_path),
+                    "td_name": td_path.name,
+                    "source": source,
+                    "attempt": attempt,
+                },
+            )
+            return
+        except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
+            last_error = e
+            if attempt + 1 < attempts:
+                # Tolerated failure path: expose it to triage but keep the
+                # timeline alive so the retry has a chance.
+                reachable(
+                    "td_runner saw tolerated failure",
+                    {
+                        "td_path": str(td_path),
+                        "td_name": td_path.name,
+                        "source": source,
+                        "attempt": attempt,
+                        "error_class": type(e).__name__,
+                        "returncode": getattr(e, "returncode", None),
+                    },
+                )
+
+    assert last_error is not None
+    raise last_error
diff --git a/misc/python/materialize/antithesis/testdrive_config.py b/misc/python/materialize/antithesis/testdrive_config.py
new file mode 100644
index 0000000000000..7a8d98a98be4f
--- /dev/null
+++ b/misc/python/materialize/antithesis/testdrive_config.py
@@ -0,0 +1,95 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+"""Shared `testdrive` invocation config for Antithesis workload helpers.
+
+Both `materialize.antithesis.upsert_sources` (the upsert-source prototype) and
+`materialize.antithesis.td_runner` (the generic .td file runner) construct the
+same set of `testdrive` arguments from the environment. This module owns that
+construction so the two paths can't drift.
+
+The corresponding env vars are populated by the test-driver service block in
+`antithesis/configs/<scenario>/mzcompose.py`.
+"""
+
+from __future__ import annotations
+
+import os
+import shlex
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class TestdriveConfig:
+    """Connection details and default flags for invoking `testdrive`.
+
+    Constructed via `TestdriveConfig.from_env()` so the test-driver container
+    can override defaults through its environment block without touching code.
+    """
+
+    testdrive_bin: str
+    materialize_url: str
+    materialize_internal_url: str
+    kafka_addr: str
+    schema_registry_url: str
+    default_timeout: str
+    seed: str | None
+    extra_args: tuple[str, ...]
+
+    @classmethod
+    def from_env(cls) -> TestdriveConfig:
+        return cls(
+            testdrive_bin=os.environ.get("MZ_TESTDRIVE", "testdrive"),
+            materialize_url=os.environ.get(
+                "MZ_MATERIALIZE_URL",
+                "postgres://materialize@materialized:6875",
+            ),
+            materialize_internal_url=os.environ.get(
+                "MZ_MATERIALIZE_INTERNAL_URL",
+                "postgres://mz_system@materialized:6877",
+            ),
+            kafka_addr=os.environ.get("MZ_KAFKA_ADDR", "kafka:9092"),
+            schema_registry_url=os.environ.get(
+                "MZ_SCHEMA_REGISTRY_URL", "http://schema-registry:8081"
+            ),
+            default_timeout=os.environ.get("MZ_TESTDRIVE_DEFAULT_TIMEOUT", "60s"),
+            seed=os.environ.get("MZ_ANTITHESIS_SEED"),
+            extra_args=tuple(
+                shlex.split(os.environ.get("MZ_TESTDRIVE_EXTRA_ARGS", ""))
+            ),
+        )
+
+    def base_command(self, *, no_reset: bool, source: str) -> list[str]:
+        """Build the testdrive argv prefix shared by all callers.
+
+        Caller is expected to append per-call args (e.g. --var=k=v) and the
+        .td path or --no-script-input usage as appropriate.
+
+        `--backoff-factor=1` keeps testdrive's retry budget short; the
+        reasoning is that under active Antithesis fault injection we want
+        testdrive to surface failures quickly instead of masking them with
+        long internal retries. The Antithesis-side wrapper decides whether
+        and how to retry.
+        """
+        cmd = [
+            self.testdrive_bin,
+            f"--kafka-addr={self.kafka_addr}",
+            f"--schema-registry-url={self.schema_registry_url}",
+            f"--materialize-url={self.materialize_url}",
+            f"--materialize-internal-url={self.materialize_internal_url}",
+            f"--default-timeout={self.default_timeout}",
+            "--backoff-factor=1",
+            f"--source={source}",
+            *self.extra_args,
+        ]
+        if self.seed is not None:
+            cmd.append(f"--seed={self.seed}")
+        if no_reset:
+            cmd.append("--no-reset")
+        return cmd
diff --git a/misc/python/materialize/antithesis/testdrive_corpus.py b/misc/python/materialize/antithesis/testdrive_corpus.py
new file mode 100644
index 0000000000000..1a0b4fac95504
--- /dev/null
+++ b/misc/python/materialize/antithesis/testdrive_corpus.py
@@ -0,0 +1,212 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+"""Curated lists of `test/testdrive/*.td` files driven by the Antithesis
+Test Composer area templates.
+
+Each list is the corpus a single area template picks from at random — see
+`antithesis/test/v1/testdrive_<area>/singleton_driver_random_<area>`. The
+template uses the Antithesis SDK's `random_choice` so the platform's
+randomness, replay, and search interact with the selection.
+
+Why area lists rather than one flat list:
+
+  * Antithesis groups assertions and triage signal by template; per-area
+    buckets give us "Kafka-side regressions vs SQL-side regressions vs
+    storage-recovery regressions" attribution at the platform level.
+  * The `source` field passed to testdrive captures the specific file inside
+    each timeline, so per-.td detail still reaches triage when needed.
+  * Adding a new file is a one-line change here. No new wrapper script.
+
+Why curated rather than auto-discover everything:
+
+  * `test/testdrive` has ~278 files; ~257 are technically base-compatible
+    (no external services), but several rely on RESET semantics, depend on
+    specific runtime knobs, or are intentionally Nightly-only. Curating
+    keeps the harness's signal-to-noise ratio high during early runs and
+    lets us expand once we trust each batch.
+
+Adding a file:
+
+  1. Skim the .td; confirm it works against `materialized + redpanda`
+     without external services and finishes in ~tens of seconds with
+     `--no-reset`.
+  2. Append it to the relevant `BASE_*` list below.
+  3. Rebuild the test-driver image so the new file lands at
+     `/opt/materialize/td/testdrive/<name>.td`.
+
+The same pattern extends to scenarios with extra services — those will
+grow new constants here (e.g. `PG_CDC`, `MYSQL_CDC`) when their scenarios
+land.
+"""
+
+from __future__ import annotations
+
+# ---------------------------------------------------------------------------
+# Files we deliberately exclude from every list.
+# ---------------------------------------------------------------------------
+
+# Mirrors the `--slow`-gated skip list inside test/testdrive/mzcompose.py
+# plus a few that are known to be flaky or to require non-default
+# configuration. Anything in this set should never appear in a BASE_* list.
+EXCLUDED: frozenset[str] = frozenset(
+    [
+        "explain-pushdown.td",  # nightly-only
+        "fivetran-destination.td",  # needs fivetran service
+        "kafka-upsert-sources.td",  # very long; covered by upsert_sources prototype
+        "materialized-view-refresh-options.td",  # nightly-only
+        "upsert-source-race.td",  # known flaky
+    ]
+)
+
+
+# ---------------------------------------------------------------------------
+# `base` scenario corpus — runs against materialized + redpanda only.
+#
+# Each list targets a single Antithesis Test Composer area template. Files
+# may appear in more than one list (e.g. a recovery-flavored Kafka file in
+# both BASE_KAFKA and BASE_RECOVERY) — Antithesis picks per area, so the
+# overlap just means more chances to land on a high-value file.
+# ---------------------------------------------------------------------------
+
+# Pure SQL: tables, DDL, joins, transactions, system tables, EXPLAIN.
+# Excludes anything that creates a Kafka/load-generator source.
+BASE_SQL: list[str] = [
+    "array.td",
+    "char-varchar-multibyte.td",
+    "concurrent-ddl.td",
+    "constants.td",
+    "decimal-overflow.td",
+    "drop.td",
+    "empty-array.td",
+    "explain-timestamps.td",
+    "fetch-tail-retraction.td",
+    "fetch-timeout.td",
+    "functions.td",
+    "insert-select.td",
+    "joins.td",
+    "jsonb.td",
+    "linear-join-fuel.td",
+    "numeric.td",
+    "oid.td",
+    "runtime-errors.td",
+    "search_path.td",
+    "show_columns.td",
+    "system-functions.td",
+    "type_char_quoted.td",
+]
+
+# Kafka source/sink + Schema Registry. Covers UPSERT/NONE/DEBEZIUM
+# envelopes, Avro/JSON/Protobuf formats, multi-partition sink behavior,
+# exactly-once sinks, and the Avro resolution matrix.
+BASE_KAFKA: list[str] = [
+    "avro-decode-bad-json.td",
+    "avro-decode-multi-record.td",
+    "avro-decode-no-record.td",
+    "avro-decode-uuid.td",
+    "avro-resolution-less-columns.td",
+    "avro-resolution-more-columns.td",
+    "avro-resolution-no-publish-reader.td",
+    "avro-resolution-no-publish-writer.td",
+    "avro-resolution-types-array.td",
+    "avro-resolution-types-map.td",
+    "avro-resolution-types-records.td",
+    "avro-resolution-types.td",
+    "avro-resolution-union-concrete.td",
+    "avro-resolution-union-reader.td",
+    "avro-resolution-union-writer.td",
+    "avro-resolution-unions.td",
+    "kafka-alter-source.td",
+    "kafka-avro-sinks-defaults.td",
+    "kafka-avro-sinks.td",
+    "kafka-exactly-once-sink.td",
+    "kafka-json-sinks.td",
+    "kafka-progress.td",
+    "kafka-recreate-topic.td",
+    "kafka-sink-empty-progress-topic.td",
+    "kafka-sink-headers.td",
+    "kafka-sink-multi-partition.td",
+    "kafka-sink-statistics.td",
+    "kafka-start-offset.td",
+    "materialized-views.td",
+    "mz-kafka-sources.td",
+    "mz-sinks.td",
+    "protobuf-map.td",
+    "protobuf-recursive.td",
+    "protobuf-required.td",
+    "upsert-unordered-key.td",
+]
+
+# LOAD GENERATOR-only files: COUNTER, AUCTION, CLOCK, DATUMS sources.
+# Exercises the storage ingest control plane independently of Kafka.
+BASE_LOAD_GENERATOR: list[str] = [
+    "compute-logical-backpressure.td",
+    "dataflow-cleanup.td",
+    "get-started.td",
+    "github-6942.td",
+    "index-advice.td",
+    "index-source-stuck.td",
+    "load-generator.td",
+    "mz-support-privileges.td",
+    "partition-by.td",
+    "quickstart.td",
+    "show-subsources.td",
+    "source-statistics-view.td",
+]
+
+# Files chosen specifically because their property is "the system survives
+# and recovers from a perturbation." High signal under Antithesis fault
+# injection. Many of these intentionally overlap with BASE_KAFKA /
+# BASE_LOAD_GENERATOR — the area-template split is for triage attribution,
+# not exclusivity.
+BASE_RECOVERY: list[str] = [
+    "dataflow-cleanup.td",
+    "hydration-status.td",
+    "kafka-progress.td",
+    "kafka-recreate-topic.td",
+    "kafka-sink-empty-progress-topic.td",
+    "materialized-views.td",
+    "sequential-hydration.td",
+    "status-history.td",
+]
+
+
+# ---------------------------------------------------------------------------
+# Self-checks: run at import time so a typo in a list above fails fast
+# locally rather than waiting for a timeline to actually try the file.
+# ---------------------------------------------------------------------------
+
+
+def _validate() -> None:
+    overlap_with_excluded = (
+        set(BASE_SQL + BASE_KAFKA + BASE_LOAD_GENERATOR + BASE_RECOVERY) & EXCLUDED
+    )
+    if overlap_with_excluded:
+        raise ValueError(
+            f"BASE_* lists contain EXCLUDED entries: {sorted(overlap_with_excluded)}"
+        )
+
+    for name, lst in (
+        ("BASE_SQL", BASE_SQL),
+        ("BASE_KAFKA", BASE_KAFKA),
+        ("BASE_LOAD_GENERATOR", BASE_LOAD_GENERATOR),
+        ("BASE_RECOVERY", BASE_RECOVERY),
+    ):
+        if not lst:
+            raise ValueError(f"{name} is empty")
+        seen: set[str] = set()
+        for entry in lst:
+            if not entry.endswith(".td"):
+                raise ValueError(f"{name}: non-.td entry {entry!r}")
+            if entry in seen:
+                raise ValueError(f"{name}: duplicate entry {entry!r}")
+            seen.add(entry)
+
+
+_validate()
diff --git a/misc/python/materialize/antithesis/upsert_sources.py b/misc/python/materialize/antithesis/upsert_sources.py
new file mode 100644
index 0000000000000..deea089b9865a
--- /dev/null
+++ b/misc/python/materialize/antithesis/upsert_sources.py
@@ -0,0 +1,266 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+from __future__ import annotations
+
+import os
+import subprocess
+import time
+from collections.abc import Iterable
+from textwrap import dedent
+
+from materialize.antithesis.sdk import (
+    always,
+    random_choice,
+    random_int,
+    reachable,
+    sometimes,
+    stop_faults,
+)
+from materialize.antithesis.testdrive_config import TestdriveConfig
+
+TOPIC = "antithesis-upsert"
+SOURCE = "antithesis_upsert"
+SOURCE_TABLE = "antithesis_upsert_tbl"
+EXPECTED_TABLE = "antithesis_upsert_expected"
+CLUSTER = "antithesis_upsert_cluster"
+KAFKA_CONNECTION = "antithesis_upsert_kafka"
+
+
+def run_testdrive(
+    script: str,
+    *,
+    config: TestdriveConfig | None = None,
+    no_reset: bool = True,
+    source: str,
+    allow_failure: bool = False,
+) -> bool:
+    config = config or TestdriveConfig.from_env()
+    cmd = config.base_command(no_reset=no_reset, source=source)
+    try:
+        subprocess.run(
+            cmd,
+            input=dedent(script).lstrip(),
+            text=True,
+            check=True,
+        )
+        return True
+    except subprocess.CalledProcessError as e:
+        if not allow_failure:
+            raise
+        reachable(
+            "upsert source command saw tolerated testdrive failure",
+            {"source": source, "returncode": e.returncode},
+        )
+        return False
+
+
+def _sql_string(value: str) -> str:
+    return "'" + value.replace("'", "''") + "'"
+
+
+def _sql_string_list(values: Iterable[str]) -> str:
+    return ", ".join(_sql_string(value) for value in values)
+
+
+def _topic_name() -> str:
+    return f"testdrive-{TOPIC}-${{testdrive.seed}}"
+
+
+def setup_text_upsert_source() -> None:
+    """Create a Kafka text/text upsert source and an expected-state table."""
+
+    cluster_size = os.environ.get("MZ_ANTITHESIS_UPSERT_CLUSTER_SIZE")
+    cluster_ddl = (
+        f"> CREATE CLUSTER {CLUSTER} SIZE {_sql_string(cluster_size)};"
+        if cluster_size
+        else ""
+    )
+    cluster_clause = f"IN CLUSTER {CLUSTER}" if cluster_size else ""
+    run_testdrive(
+        f"""
+        $ set-sql-timeout duration=60s
+
+        $ kafka-create-topic topic={TOPIC} partitions=1
+
+        > CREATE CONNECTION IF NOT EXISTS {KAFKA_CONNECTION}
+          TO KAFKA (BROKER '${{testdrive.kafka-addr}}', SECURITY PROTOCOL PLAINTEXT);
+
+        > DROP TABLE IF EXISTS {EXPECTED_TABLE};
+        > DROP SOURCE IF EXISTS {SOURCE} CASCADE;
+        > DROP CLUSTER IF EXISTS {CLUSTER} CASCADE;
+
+        {cluster_ddl}
+
+        > BEGIN
+        > CREATE SOURCE {SOURCE}
+          {cluster_clause}
+          FROM KAFKA CONNECTION {KAFKA_CONNECTION} (TOPIC '{_topic_name()}');
+
+        > CREATE TABLE {SOURCE_TABLE} FROM SOURCE {SOURCE} (REFERENCE "{_topic_name()}")
+          KEY FORMAT TEXT
+          VALUE FORMAT TEXT
+          ENVELOPE UPSERT;
+        > COMMIT
+
+        > CREATE TABLE {EXPECTED_TABLE} (key text, value text);
+
+        > SELECT count(*) FROM {SOURCE_TABLE};
+        0
+        """,
+        no_reset=False,
+        source="antithesis/upsert_sources/first_create_upsert_source",
+    )
+    reachable("upsert source setup completed", {"topic": TOPIC})
+
+
+def write_random_upserts(*, allow_failure: bool) -> None:
+    batch_size = int(os.environ.get("MZ_ANTITHESIS_UPSERT_BATCH_SIZE", "8"))
+    batch_id = f"{int(time.time() * 1000)}_{random_int(1_000_000)}"
+
+    latest: dict[str, str | None] = {}
+    kafka_lines: list[str] = []
+    key_space = int(os.environ.get("MZ_ANTITHESIS_UPSERT_KEY_SPACE", "64"))
+    attempted_upserts = 0
+    attempted_deletes = 0
+
+    for i in range(batch_size):
+        key = f"k{random_int(key_space):03d}"
+        if random_choice([False, False, False, True]):
+            attempted_deletes += 1
+            latest[key] = None
+            kafka_lines.append(f"{key}:")
+        else:
+            attempted_upserts += 1
+            value = f"v_{batch_id}_{i}_{random_int(1_000_000)}"
+            latest[key] = value
+            kafka_lines.append(f"{key}:{value}")
+
+    upserts = [(key, value) for key, value in latest.items() if value is not None]
+    sql_delete_keys = list(latest)
+    expected_update = ""
+    if sql_delete_keys:
+        expected_update += (
+            f"> DELETE FROM {EXPECTED_TABLE} "
+            f"WHERE key IN ({_sql_string_list(sql_delete_keys)});\n"
+        )
+    if upserts:
+        values = ", ".join(
+            f"({_sql_string(key)}, {_sql_string(value)})" for key, value in upserts
+        )
+        expected_update += f"> INSERT INTO {EXPECTED_TABLE} VALUES {values};\n"
+
+    succeeded = run_testdrive(
+        f"""
+        $ set-sql-timeout duration=60s
+
+        $ kafka-ingest format=bytes topic={TOPIC} key-format=bytes key-terminator=:
+        {chr(10).join(kafka_lines)}
+
+        {expected_update}
+        """,
+        no_reset=True,
+        source="antithesis/upsert_sources/parallel_driver_write_upserts",
+        allow_failure=allow_failure,
+    )
+    sometimes(
+        succeeded,
+        "upsert source write driver completed at least once",
+        {
+            "batch_size": batch_size,
+            "attempted_upserts": attempted_upserts,
+            "attempted_deletes": attempted_deletes,
+            "latest_upserts": len(upserts),
+        },
+    )
+
+
+def read_stale_safe(*, allow_failure: bool) -> None:
+    succeeded = run_testdrive(
+        f"""
+        $ set-sql-timeout duration=30s
+
+        > SELECT count(*) >= 0 FROM {SOURCE_TABLE};
+        true
+
+        > SELECT count(*) >= 0 FROM {EXPECTED_TABLE};
+        true
+        """,
+        no_reset=True,
+        source="antithesis/upsert_sources/parallel_driver_read_stale_safe",
+        allow_failure=allow_failure,
+    )
+    sometimes(
+        succeeded,
+        "upsert source stale-safe read completed at least once",
+        {"source_table": SOURCE_TABLE},
+    )
+
+
+def health_check(*, allow_failure: bool) -> None:
+    succeeded = run_testdrive(
+        f"""
+        $ set-sql-timeout duration=30s
+
+        > SELECT status IN ('running', 'starting', 'stalled')
+          FROM mz_internal.mz_source_statuses
+          WHERE name = '{SOURCE_TABLE}';
+        true
+        """,
+        no_reset=True,
+        source="antithesis/upsert_sources/anytime_upsert_source_health",
+        allow_failure=allow_failure,
+    )
+    sometimes(
+        succeeded,
+        "upsert source health check completed at least once",
+        {"source_table": SOURCE_TABLE},
+    )
+
+
+def verify_expected_rows_visible(*, source: str) -> None:
+    run_testdrive(
+        f"""
+        $ set-sql-timeout duration=120s
+
+        > SELECT e.key, e.value, u.text
+          FROM {EXPECTED_TABLE} e
+          LEFT JOIN {SOURCE_TABLE} u ON e.key = u.key
+          WHERE u.key IS NULL OR u.text <> e.value;
+        """,
+        no_reset=True,
+        source=source,
+    )
+    reachable("upsert source expected rows verified", {"source": source})
+
+
+def write_sentinel_and_verify() -> None:
+    stop_faults(int(os.environ.get("MZ_ANTITHESIS_QUIET_SECONDS", "20")))
+    key = f"sentinel_{int(time.time() * 1000)}_{random_int(1_000_000)}"
+    value = f"sentinel_value_{random_int(1_000_000)}"
+    run_testdrive(
+        f"""
+        $ set-sql-timeout duration=120s
+
+        $ kafka-ingest format=bytes topic={TOPIC} key-format=bytes key-terminator=:
+        {key}:{value}
+
+        > DELETE FROM {EXPECTED_TABLE} WHERE key = {_sql_string(key)};
+        > INSERT INTO {EXPECTED_TABLE} VALUES ({_sql_string(key)}, {_sql_string(value)});
+
+        > SELECT text FROM {SOURCE_TABLE} WHERE key = {_sql_string(key)};
+        {value}
+        """,
+        no_reset=True,
+        source="antithesis/upsert_sources/eventually_upsert_source_catches_up",
+    )
+    verify_expected_rows_visible(
+        source="antithesis/upsert_sources/eventually_upsert_source_catches_up"
+    )
+    always(True, "upsert source catches up after quiet period", {"key": key})
diff --git a/misc/python/materialize/cli/mzcompose.py b/misc/python/materialize/cli/mzcompose.py
index 0e54cad23e940..29fbc8a7eea5b 100644
--- a/misc/python/materialize/cli/mzcompose.py
+++ b/misc/python/materialize/cli/mzcompose.py
@@ -163,10 +163,19 @@ def main(argv: list[str]) -> None:
     )
     shtab.add_argument_to(completion_parser, "shell", parent=parser)
 
-    # shtab can't handle Enum choices (indexes with [0]). Convert to strings.
+    # shtab can't handle an Enum class as `choices` because it tries to index
+    # the choices container with `[0]` to sample a value, and Enum classes
+    # don't support integer subscripting.
+    #
+    # The fix is to expose `choices` as an iterable of Enum *members*, not a
+    # list of names or values. Argparse coerces input via `type=Enum` to a
+    # member, and then performs the choice check via `member in choices`. If
+    # `choices` is a list of strings, that check fails because Enum members
+    # don't compare equal to strings; if it's a list of members, both shtab
+    # and argparse are happy.
     for action in parser._actions:  # noqa: SLF001
         if isinstance(action.choices, type) and issubclass(action.choices, enum.Enum):
-            action.choices = [e.name for e in action.choices]
+            action.choices = list(action.choices)
 
     args = parser.parse_args(argv)
     if args.file:
diff --git a/misc/python/materialize/mzbuild.py b/misc/python/materialize/mzbuild.py
index f653b84abc4a9..52ebf70c2f89a 100644
--- a/misc/python/materialize/mzbuild.py
+++ b/misc/python/materialize/mzbuild.py
@@ -471,13 +471,24 @@ def __init__(self, rd: RepositoryDetails, path: Path, config: dict[str, Any]):
 
     def run(self, prep: Any) -> None:
         super().run(prep)
-        for src in self.inputs():
-            dst = self.path / self.destination / src
+        source_prefix = f"{self.source}/"
+        for input in self.inputs():
+            # `inputs()` reports paths relative to the repository root so the
+            # mzbuild fingerprinter can locate them. Strip the source prefix
+            # before resolving the destination inside the build context.
+            assert input.startswith(source_prefix), (
+                f"Copy pre-image input {input!r} does not start with source "
+                f"{source_prefix!r}"
+            )
+            relative = input[len(source_prefix) :]
+            dst = self.path / self.destination / relative
             dst.parent.mkdir(parents=True, exist_ok=True)
-            shutil.copy(self.rd.root / self.source / src, dst)
+            shutil.copy(self.rd.root / input, dst)
 
     def inputs(self) -> set[str]:
-        return set(git.expand_globs(self.rd.root / self.source, self.matching))
+        # Paths are relative to the repository root so that
+        # `ResolvedImage.fingerprint` can resolve them via `rd.root / path`.
+        return set(git.expand_globs(self.rd.root, f"{self.source}/{self.matching}"))
 
 
 class CargoPreImage(PreImage):
diff --git a/misc/python/materialize/mzcompose/service.py b/misc/python/materialize/mzcompose/service.py
index 3d5d4e4c38b0c..42c71ca690712 100644
--- a/misc/python/materialize/mzcompose/service.py
+++ b/misc/python/materialize/mzcompose/service.py
@@ -103,6 +103,13 @@ class ServiceConfig(TypedDict, total=False):
     By default, the container's ID is used as the hostname.
     """
 
+    container_name: str
+    """A pinned name for the container.
+
+    Required by Antithesis so log lines and fault attribution can use the
+    service name; otherwise Docker Compose synthesizes a per-project name.
+    """
+
     extra_hosts: list[str]
     """Additional hostname mappings."""