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."""