diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 493f67a..63879be 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -18,6 +18,12 @@ jobs:
       - name: Check out the repository
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
 
+      - name: Set up Erlang and Elixir for staging scripts
+        uses: erlef/setup-beam@54075bcc5e249e4758d363f27d099f55d843f124 # v1.24.1
+        with:
+          version-file: .tool-versions
+          version-type: strict
+
       - name: Set up Node.js for the packaged browser pilot
         uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0
         with:
@@ -32,7 +38,7 @@ jobs:
         run: docker build --build-arg REVISION="$GITHUB_SHA" -t synixir:ci .
 
       - name: Rehearse HTTPS, native libraries, replacement and database recovery
-        run: python3 scripts/staging.py rehearse --image synixir:ci --output /tmp/synixir-release-pilot.json
+        run: elixir scripts/staging.exs rehearse --image synixir:ci --output /tmp/synixir-release-pilot.json
 
   test:
     name: Format, compile, and test
@@ -78,6 +84,9 @@ jobs:
       - name: Run tests
         run: mix test
 
+      - name: Test standalone operations scripts
+        run: elixir scripts/operations_test.exs
+
       - name: Set up Node.js for the browser example
         uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0
         with:
@@ -104,10 +113,10 @@ jobs:
         run: npm test
 
       - name: Verify backup restoration and concurrent limits in isolated databases
-        run: python3 scripts/operations/database.py --docker-container '${{ job.services.postgres.id }}' drill --output /tmp/synixir-restore.json
+        run: elixir scripts/operations/database.exs --docker-container '${{ job.services.postgres.id }}' drill --output /tmp/synixir-restore.json
 
       - name: Exercise SDK load, reconnects, large transfers and persisted replay
-        run: python3 scripts/operations/database.py --docker-container '${{ job.services.postgres.id }}' load --output /tmp/synixir-load.json
+        run: elixir scripts/operations/database.exs --docker-container '${{ job.services.postgres.id }}' load --output /tmp/synixir-load.json
 
       - name: Validate Prometheus configuration, alert behavior and actual metrics
         env:
diff --git a/README.md b/README.md
index c3d10d4..3b23156 100644
--- a/README.md
+++ b/README.md
@@ -35,13 +35,15 @@ require separate setup.
 
 ## Try local Docker staging
 
-Start Docker or OrbStack. With Docker Compose v2, Python 3.11 or newer, and
-OpenSSL installed, run these commands from the repository for the first setup:
+Start Docker or OrbStack. Install Docker Compose v2, OpenSSL, and the Elixir and
+Erlang versions in `.tool-versions`. The standalone scripts require Elixir 1.18+
+and Erlang/OTP 27+ with no Mix dependencies. Run these commands from the repository
+for the first setup:
 
 ```sh
-python3 scripts/staging.py init
-python3 scripts/staging.py build
-python3 scripts/staging.py up
+elixir scripts/staging.exs init
+elixir scripts/staging.exs build
+elixir scripts/staging.exs up
 ```
 
 Open [https://localhost:8443](https://localhost:8443), accept the local self-signed
@@ -49,8 +51,8 @@ certificate, and create an account and room. The release includes all browser
 examples and uses its own PostgreSQL volume and private Prometheus collector.
 Docker builds Elixir, Erlang, and the frontend inside the image.
 
-For an existing installation, run `python3 scripts/staging.py up` to start it.
-Use `python3 scripts/staging.py stop` to stop it while retaining the database.
+For an existing installation, run `elixir scripts/staging.exs up` to start it.
+Use `elixir scripts/staging.exs stop` to stop it while retaining the database.
 Initialization runs only once; keep the private `.local/staging` configuration
 with its database volume. The [staging runbook](docs/deployment.md) covers the
 browser pilot, certificate renewal, backups, and upgrades.
diff --git a/deploy/compose.yaml b/deploy/compose.yaml
index 19124b1..652a5de 100644
--- a/deploy/compose.yaml
+++ b/deploy/compose.yaml
@@ -1,4 +1,4 @@
-name: ${SYNIXIR_STAGE_PROJECT:?Initialize staging with scripts/staging.py init}
+name: ${SYNIXIR_STAGE_PROJECT:?Initialize staging with scripts/staging.exs init}
 
 x-release: &release
   image: ${SYNIXIR_IMAGE:?Set the locally built release image}
diff --git a/docs/deployment.md b/docs/deployment.md
index 36cda0a..0057c45 100644
--- a/docs/deployment.md
+++ b/docs/deployment.md
@@ -6,15 +6,18 @@ HTTPS gateway and private Prometheus collector. It runs one application node.
 
 ## Start staging
 
-Requirements: Docker with Compose v2, Python 3.11 or newer, OpenSSL, and Node.js
-matching `examples/collaboration/.nvmrc` for the browser pilot. Docker builds Elixir,
-Erlang and the frontend inside the image. Run these commands from the repository:
+Requirements: Docker with Compose v2, OpenSSL, and Elixir 1.18+ with Erlang/OTP 27+.
+Use the versions pinned in `.tool-versions`. The standalone staging scripts use
+only the Elixir and Erlang standard libraries, so they need no `mix deps.get`.
+The browser pilot also needs Node.js matching `examples/collaboration/.nvmrc`.
+Docker builds the application and frontend inside the image. Run these commands
+from the repository:
 
 ```sh
-python3 scripts/staging.py init
-python3 scripts/staging.py build
-python3 scripts/staging.py up
-python3 scripts/staging.py status
+elixir scripts/staging.exs init
+elixir scripts/staging.exs build
+elixir scripts/staging.exs up
+elixir scripts/staging.exs status
 ```
 
 Open [local staging](https://localhost:8443) and create an account. The SDK settings
@@ -41,7 +44,7 @@ uses the `synixir-staging_database` volume. It does not use the development data
 or ports 4000, 5173 or 5432. To select another port or image at initialization:
 
 ```sh
-python3 scripts/staging.py init --port 8445 --image synixir:staging-v1
+elixir scripts/staging.exs init --port 8445 --image synixir:staging-v1
 ```
 
 Use `--directory /absolute/private/path` before the subcommand to select a saved
@@ -102,7 +105,7 @@ Install the browser pilot dependencies once:
 ```sh
 npm ci
 npx playwright install --only-shell chromium
-python3 scripts/staging.py pilot
+elixir scripts/staging.exs pilot
 ```
 
 The pilot creates owner, editor and viewer accounts. It checks shared editing,
@@ -118,8 +121,8 @@ The private `pilot.json` retains fixture credentials and browser sessions.
 pilot after an upgrade:
 
 ```sh
-python3 scripts/staging.py check-pilot
-python3 scripts/staging.py native-check
+elixir scripts/staging.exs check-pilot
+elixir scripts/staging.exs native-check
 ```
 
 The pilot's document and memberships are fixtures. Editing them changes the
@@ -127,7 +130,7 @@ expected results. To run the entire rehearsal again without touching this
 installation, CI and local development use a fresh, randomly named project:
 
 ```sh
-python3 scripts/staging.py rehearse --image synixir:staging \
+elixir scripts/staging.exs rehearse --image synixir:staging \
   --output /tmp/synixir-release-pilot.json
 ```
 
@@ -138,8 +141,8 @@ for each run. It never deletes the persistent staging volume.
 ## Backups and replacement
 
 ```sh
-python3 scripts/staging.py backup
-python3 scripts/staging.py verify /absolute/path/to/backup.dump
+elixir scripts/staging.exs backup
+elixir scripts/staging.exs verify /absolute/path/to/backup.dump
 ```
 
 Backups are full PostgreSQL custom archives with a SHA-256 sidecar. Verification
@@ -153,8 +156,8 @@ Build a distinct image tag for a candidate, then replace staging:
 
 ```sh
 docker build --build-arg REVISION="$(git rev-parse HEAD)" -t synixir:staging-v2 .
-python3 scripts/staging.py upgrade --image synixir:staging-v2
-python3 scripts/staging.py check-pilot
+elixir scripts/staging.exs upgrade --image synixir:staging-v2
+elixir scripts/staging.exs check-pilot
 ```
 
 For an uncommitted build, append `-dirty` to its revision label, as the `build`
@@ -185,8 +188,8 @@ arbitrary future migration compatibility.
 ## Stop and scope
 
 ```sh
-python3 scripts/staging.py stop
-python3 scripts/staging.py up
+elixir scripts/staging.exs stop
+elixir scripts/staging.exs up
 ```
 
 Stopping retains the database volume, credentials and backups. Do not remove the
diff --git a/docs/operations.md b/docs/operations.md
index 184b293..8986f0b 100644
--- a/docs/operations.md
+++ b/docs/operations.md
@@ -127,15 +127,19 @@ reachability. CI also validates an actual application scrape using `check metric
 
 ## Back up and verify
 
-Use Python 3.11 or later and PostgreSQL 17 client tools. The examples use the
-repository's PostgreSQL container, `synixir-db-1`. `--docker-container` runs tools
+Use Elixir 1.18+ with Erlang/OTP 27+ and PostgreSQL 17 client tools. The examples
+use the repository's PostgreSQL container, `synixir-db-1`. `--docker-container` runs tools
 inside that specific container against its local database. Omit it to use native
 tools with libpq's `PGHOST`, `PGPORT`, `PGUSER`, `PGPASSFILE` or `PGPASSWORD`.
 
+The CLI uses only standard libraries. Restore verification, drills, and load
+checks also run the application through Mix and require its dependencies.
+Run `elixir scripts/operations_test.exs` to test the standalone script support.
+
 ```sh
-python3 scripts/operations/database.py --docker-container synixir-db-1 backup --database synixir_dev --output /secure/backups/synixir.dump
-python3 scripts/operations/database.py --docker-container synixir-db-1 verify /secure/backups/synixir.dump
-python3 scripts/operations/database.py --docker-container synixir-db-1 drill --output /tmp/synixir-restore.json
+elixir scripts/operations/database.exs --docker-container synixir-db-1 backup --database synixir_dev --output /secure/backups/synixir.dump
+elixir scripts/operations/database.exs --docker-container synixir-db-1 verify /secure/backups/synixir.dump
+elixir scripts/operations/database.exs --docker-container synixir-db-1 drill --output /tmp/synixir-restore.json
 ```
 
 Create the output directory first and choose a new filename on each run. Backups
@@ -186,7 +190,7 @@ external storage policy.
 ## Load check
 
 ```sh
-python3 scripts/operations/database.py --docker-container synixir-db-1 load --clients 12 --rooms 3 --writes 100 --output /tmp/synixir-load.json
+elixir scripts/operations/database.exs --docker-container synixir-db-1 load --clients 12 --rooms 3 --writes 100 --output /tmp/synixir-load.json
 docker run --rm -i --network none --entrypoint promtool prom/prometheus:v3.5.5 check metrics < /tmp/synixir-load.json.prom
 ```
 
diff --git a/scripts/operations/database.exs b/scripts/operations/database.exs
new file mode 100644
index 0000000..f2e537a
--- /dev/null
+++ b/scripts/operations/database.exs
@@ -0,0 +1,79 @@
+#!/usr/bin/env elixir
+Code.require_file("../support/database.exs", __DIR__)
+
+defmodule Synixir.Script.DatabaseCLI do
+  alias Synixir.Script, as: S
+  alias Synixir.Script.Database
+
+  @usage """
+  Usage: elixir scripts/operations/database.exs [--docker-container NAME] COMMAND
+    backup --database NAME --output PATH
+    verify ARCHIVE
+    drill --output PATH
+    load --output PATH [--clients 12] [--rooms 3] [--writes 20] [--port 4012]
+  """
+
+  def main(argv) do
+    case S.parse(
+           argv,
+           [
+             docker_container: :string,
+             database: :string,
+             output: :string,
+             clients: :integer,
+             rooms: :integer,
+             writes: :integer,
+             port: :integer
+           ],
+           ~w(backup verify drill load),
+           @usage
+         ) do
+      :help -> :ok
+      {options, args} -> execute(options, args)
+    end
+  end
+
+  defp execute(options, args) do
+    container = options[:docker_container]
+    output = if options[:output], do: options[:output] |> Path.expand() |> S.unused!()
+
+    result =
+      case args do
+        ["backup"] ->
+          Database.backup(
+            container,
+            S.required(options, :database),
+            output || S.required(options, :output)
+          )
+
+        ["verify", archive] ->
+          Database.verify(container, Path.expand(archive))
+
+        ["drill"] ->
+          S.required(options, :output)
+          result = Database.drill(container)
+          S.private_json(output, result)
+          result
+
+        ["load"] ->
+          S.required(options, :output)
+          S.unused!(output <> ".prom")
+
+          Database.load(
+            container,
+            options[:clients] || 12,
+            options[:rooms] || 3,
+            options[:writes] || 20,
+            options[:port] || 4012,
+            output
+          )
+
+        _ ->
+          raise ArgumentError, @usage
+      end
+
+    S.print(result)
+  end
+end
+
+Synixir.Script.cli(fn -> Synixir.Script.DatabaseCLI.main(System.argv()) end)
diff --git a/scripts/operations/database.py b/scripts/operations/database.py
deleted file mode 100644
index 70003f3..0000000
--- a/scripts/operations/database.py
+++ /dev/null
@@ -1,229 +0,0 @@
-#!/usr/bin/env python3
-"""Full PostgreSQL backups and isolated restore/load drills. Python standard library only."""
-import argparse
-import contextlib
-import hashlib
-import json
-import os
-from pathlib import Path
-import re
-import secrets
-import signal
-import socket
-import subprocess
-import sys
-import tempfile
-import time
-from datetime import datetime, timezone
-from urllib.request import urlopen
-
-ROOT = Path(__file__).resolve().parents[2]
-ISOLATED = re.compile(r"synixir_(load|drill|restore)_[a-f0-9]{12}\Z")
-
-
-def run(args, **kwargs):
-    result = subprocess.run(args, cwd=ROOT, check=False, timeout=300, **kwargs)
-    if result.returncode:
-        for stream in [result.stdout, result.stderr]:
-            if isinstance(stream, bytes):
-                print(stream.decode(errors="replace")[-6000:], file=sys.stderr)
-        raise RuntimeError(f"{args[0]} failed (exit {result.returncode})")
-    return result
-
-
-def private_json(path, value):
-    with open(os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w") as out:
-        json.dump(value, out, indent=2)
-        out.write("\n")
-
-
-class Database:
-    def __init__(self, container):
-        self.container = container
-
-    def command(self, tool, *args):
-        # Container tools connect to their own PostgreSQL via its local socket.
-        # Native tools follow libpq's PG* environment, including PGPASSFILE.
-        prefix = ["docker", "exec", "-i", self.container] if self.container else []
-        user = ["--username", os.environ.get("PGUSER", "postgres")] if self.container and "--version" not in args else []
-        return prefix + [tool] + user + list(args)
-
-    def sql(self, database, sql):
-        return run(self.command("psql", "--no-psqlrc", "--set", "ON_ERROR_STOP=1",
-                                "--dbname", database, "--tuples-only", "--no-align"),
-                   input=sql.encode(), stdout=subprocess.PIPE).stdout.decode().strip()
-
-    @contextlib.contextmanager
-    def fresh(self, kind):
-        name = f"synixir_{kind}_{secrets.token_hex(6)}"
-        assert ISOLATED.fullmatch(name)
-        # Only a successfully created name enters cleanup. Never adopt an existing DB.
-        self.sql("postgres", f'CREATE DATABASE "{name}" TEMPLATE template0;')
-        try:
-            yield name
-        finally:
-            self.sql("postgres", f'DROP DATABASE "{name}" WITH (FORCE);')
-
-    def backup(self, database, archive):
-        if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,62}", database):
-            raise ValueError("Pass a plain database name, not a connection string")
-        if archive.exists() or Path(str(archive) + ".json").exists():
-            raise ValueError("Backup output already exists")
-        started = datetime.now(timezone.utc).isoformat()
-        migrations = self.sql(database, "SELECT version FROM schema_migrations ORDER BY version").splitlines()
-        version = run(self.command("pg_dump", "--version"), stdout=subprocess.PIPE).stdout.decode().strip()
-        with open(os.open(archive, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "wb") as out:
-            try:
-                run(self.command("pg_dump", "--format=custom", "--dbname", database), stdout=out)
-            except BaseException:
-                archive.unlink(missing_ok=True)
-                raise
-        with archive.open("rb") as data:
-            digest = hashlib.file_digest(data, "sha256").hexdigest()
-        metadata = {"started_at": started, "finished_at": datetime.now(timezone.utc).isoformat(),
-                    "database": database, "tool": version, "migrations_before_dump": migrations,
-                    "bytes": archive.stat().st_size,
-                    "sha256": digest}
-        private_json(str(archive) + ".json", metadata)
-        return metadata
-
-    def restore(self, archive, target, role=None):
-        assert ISOLATED.fullmatch(target)
-        metadata = json.loads(Path(str(archive) + ".json").read_text())
-        with archive.open("rb") as data:
-            if hashlib.file_digest(data, "sha256").hexdigest() != metadata["sha256"]:
-                raise ValueError("Archive checksum mismatch")
-            data.seek(0)
-            role_args = ["--role", role] if role else []
-            run(self.command("pg_restore", "--single-transaction", "--exit-on-error",
-                             "--no-owner", "--no-privileges", "--dbname", target, *role_args), stdin=data)
-
-
-def app_env(database):
-    assert ISOLATED.fullmatch(database)
-    env = dict(os.environ, MIX_ENV="test", SYNIXIR_OPERATIONS_DATABASE=database)
-    for key in ["PHX_SERVER", "SYNIXIR_BROWSER_TEST", "DATABASE_URL", "MIX_TEST_PARTITION"]:
-        env.pop(key, None)
-    return env
-
-
-def mix(database, *args):
-    result = run(["mix", *args], env=app_env(database), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    output = result.stdout.decode()
-    for line in output.splitlines():
-        if line.startswith("OPERATIONS_RESULT="):
-            return json.loads(line.removeprefix("OPERATIONS_RESULT="))
-    return None
-
-
-def check(database, *args):
-    return mix(database, "run", "scripts/operations/restore.exs", *args)
-
-
-def verify(db, archive, fixture=None):
-    started = time.monotonic()
-    with db.fresh("restore") as target:
-        db.restore(archive, target)
-        result = check(target, "drill-verify", str(fixture)) if fixture else check(target, "verify")
-    return dict(result, restore_and_check_seconds=round(time.monotonic() - started, 3))
-
-
-def drill(db):
-    with tempfile.TemporaryDirectory(prefix="synixir-drill-") as directory, db.fresh("drill") as source:
-        mix(source, "ecto.migrate", "--quiet")
-        mix(source, "run", "scripts/operations/limits.exs")
-        fixture = Path(directory) / "expectations.json"
-        private_json(fixture, {})
-        check(source, "seed", str(fixture))
-        archive = Path(directory) / "backup.dump"
-        backup = db.backup(source, archive)
-        return dict(verify(db, archive, fixture), archive_bytes=backup["bytes"],
-                    recorded_at=backup["finished_at"], backup_tool=backup["tool"],
-                    isolated_limits_checks=True)
-
-
-def load(db, clients, rooms, writes, port, output):
-    if not (1 <= clients <= 100 and 1 <= rooms <= clients and 1 <= writes <= 1000):
-        raise ValueError("Use 1–100 clients, 1–clients rooms, and 1–1000 writes per client")
-    if not 1024 <= port <= 65535 or port in [4000, 4010, 5173, 5174]:
-        raise ValueError("Choose an unused unprivileged port separate from development/browser fixtures")
-    with socket.socket() as probe:
-        probe.bind(("127.0.0.1", port))
-    with db.fresh("load") as database, tempfile.TemporaryFile() as logs:
-        mix(database, "ecto.migrate", "--quiet")
-        env = dict(app_env(database), PHX_SERVER="true", PORT=str(port),
-                   SYNIXIR_METRICS_TOKEN=secrets.token_hex(32))
-        child = subprocess.Popen(["mix", "phx.server"], cwd=ROOT, env=env,
-                                 stdout=logs, stderr=logs, start_new_session=True)
-        try:
-            deadline = time.monotonic() + 30
-            while True:
-                if child.poll() is not None or time.monotonic() > deadline:
-                    raise RuntimeError("Isolated load server did not start")
-                try:
-                    with urlopen(f"http://127.0.0.1:{port}/health/ready", timeout=1) as response:
-                        if response.status == 200:
-                            break
-                except OSError:
-                    time.sleep(0.1)
-            result = run(["node", "scripts/operations/load.js", str(port), str(clients),
-                          str(rooms), str(writes)], env=env, stdout=subprocess.PIPE)
-            report = json.loads(result.stdout)
-            metrics = report.pop("metrics_text")
-        except BaseException:
-            logs.seek(0, os.SEEK_END)
-            logs.seek(max(0, logs.tell() - 6000))
-            print(logs.read().decode(errors="replace"), file=sys.stderr)
-            raise
-        finally:
-            if child.poll() is None:
-                os.killpg(child.pid, signal.SIGTERM)
-                try:
-                    child.wait(timeout=10)
-                except subprocess.TimeoutExpired:
-                    os.killpg(child.pid, signal.SIGKILL)
-                    child.wait()
-        report["stored_documents"] = check(database, "verify")
-        with open(os.open(str(output) + ".prom", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w") as scrape:
-            scrape.write(metrics)
-        private_json(output, report)
-        return report
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--docker-container", help="Use this PostgreSQL container's local tools")
-    sub = parser.add_subparsers(dest="command", required=True)
-    backup = sub.add_parser("backup")
-    backup.add_argument("--database", required=True)
-    backup.add_argument("--output", type=Path, required=True)
-    restore = sub.add_parser("verify")
-    restore.add_argument("archive", type=Path)
-    for name in ["drill", "load"]:
-        command = sub.add_parser(name)
-        command.add_argument("--output", type=Path, required=True)
-        if name == "load":
-            command.add_argument("--clients", type=int, default=12)
-            command.add_argument("--rooms", type=int, default=3)
-            command.add_argument("--writes", type=int, default=20)
-            command.add_argument("--port", type=int, default=4012)
-    args = parser.parse_args()
-    db = Database(args.docker_container)
-    if getattr(args, "output", None) and args.output.exists():
-        parser.error("Output exists; choose a new path")
-    if args.command == "load" and Path(str(args.output) + ".prom").exists():
-        parser.error("Metrics output exists; choose a new path")
-    if args.command == "backup":
-        result = db.backup(args.database, args.output)
-    elif args.command == "verify":
-        result = verify(db, args.archive)
-    elif args.command == "drill":
-        result = drill(db)
-        private_json(args.output, result)
-    else:
-        result = load(db, args.clients, args.rooms, args.writes, args.port, args.output)
-    print(json.dumps(result, indent=2))
-
-
-if __name__ == "__main__":
-    main()
diff --git a/scripts/operations/load.js b/scripts/operations/load.js
index b26823d..a004453 100644
--- a/scripts/operations/load.js
+++ b/scripts/operations/load.js
@@ -1,4 +1,4 @@
-// Invoked by database.py, which owns the isolated database and server lifecycle.
+// Invoked by database.exs, which owns the isolated database and server lifecycle.
 import assert from "node:assert/strict";
 import { performance } from "node:perf_hooks";
 import { setTimeout as delay } from "node:timers/promises";
diff --git a/scripts/operations_test.exs b/scripts/operations_test.exs
new file mode 100644
index 0000000..67f3c46
--- /dev/null
+++ b/scripts/operations_test.exs
@@ -0,0 +1,176 @@
+Code.require_file("support/staging.exs", __DIR__)
+ExUnit.start()
+
+defmodule Synixir.ScriptTest do
+  use ExUnit.Case, async: false
+  alias Synixir.Script, as: S
+  alias Synixir.Script.{Database, Stage}
+
+  test "structured command results can follow ordinary application logs" do
+    assert S.result("starting app\nOPERATIONS_RESULT={\"ok\":true}\n", "OPERATIONS_RESULT=") ==
+             %{"ok" => true}
+
+    assert_raise RuntimeError, "Command did not return OPERATIONS_RESULT=", fn ->
+      S.result("starting app\n", "OPERATIONS_RESULT=")
+    end
+  end
+
+  test "HTTPS trusts the saved self-signed certificate and rejects other certificates and hosts" do
+    S.temporary(fn directory ->
+      for name <- ["trusted", "other"] do
+        S.run([
+          "openssl",
+          "req",
+          "-x509",
+          "-newkey",
+          "rsa:2048",
+          "-nodes",
+          "-days",
+          "1",
+          "-keyout",
+          Path.join(directory, name <> ".key"),
+          "-out",
+          Path.join(directory, name <> ".crt"),
+          "-subj",
+          "/CN=localhost",
+          "-addext",
+          "subjectAltName=DNS:localhost"
+        ])
+      end
+
+      {:ok, probe} = :gen_tcp.listen(0, ip: {127, 0, 0, 1})
+      {:ok, port} = :inet.port(probe)
+      :gen_tcp.close(probe)
+      trusted = Path.join(directory, "trusted.crt")
+
+      child =
+        S.start([
+          "openssl",
+          "s_server",
+          "-accept",
+          "127.0.0.1:#{port}",
+          "-www",
+          "-cert",
+          trusted,
+          "-key",
+          Path.join(directory, "trusted.key")
+        ])
+
+      try do
+        S.eventually(
+          fn -> S.http_status("https://localhost:#{port}", cacert: trusted) == 200 end,
+          "TLS fixture did not start",
+          5000
+        )
+
+        assert is_nil(
+                 S.http_status("https://localhost:#{port}",
+                   cacert: Path.join(directory, "other.crt")
+                 )
+               )
+
+        assert is_nil(S.http_status("https://127.0.0.1:#{port}", cacert: trusted))
+      after
+        S.stop(child)
+      end
+    end)
+  end
+
+  test "commands preserve literal arguments, binary streams, and environment overrides" do
+    S.temporary(fn directory ->
+      input = Path.join(directory, "input with spaces")
+      output = Path.join(directory, "output with spaces")
+      bytes = <<0, 255, 10, 13, 0, 128>>
+      S.private_file(input, bytes)
+      S.private_file(output, "")
+      S.run(["cat"], stdin: input, stdout: output)
+      assert File.read!(output) == bytes
+
+      assert S.run(["printf", "%s", "$(echo wrong) `echo wrong` ' \" $HOME"]) ==
+               "$(echo wrong) `echo wrong` ' \" $HOME"
+
+      assert S.run(["sh", "-c", "printf '%s' \"$SYNIXIR_SCRIPT_TEST\""],
+               env: [{"SYNIXIR_SCRIPT_TEST", "literal value"}]
+             ) == "literal value"
+
+      assert S.run(["sh", "-c", "printf '%s' \"${HOME-unset}\""], env: [{"HOME", nil}]) == "unset"
+    end)
+  end
+
+  test "command failures propagate and timeouts stop children" do
+    assert_raise RuntimeError, "sh exited with 7", fn -> S.run(["sh", "-c", "exit 7"]) end
+
+    S.temporary(fn directory ->
+      marker = Path.join(directory, "child survived")
+
+      assert_raise RuntimeError, "sh timed out", fn ->
+        S.run(["sh", "-c", "(sleep 1; touch \"$1\") & wait", "test", marker], timeout: 100)
+      end
+
+      Process.sleep(1100)
+      refute File.exists?(marker)
+    end)
+  end
+
+  test "private files reject replacement and private temporary directories clean up on error" do
+    directory =
+      S.temporary(fn directory ->
+        path = Path.join(directory, "secret.json")
+        S.private_json(path, %{secret: "value"})
+        assert Bitwise.band(File.stat!(path).mode, 0o777) == 0o600
+        assert Bitwise.band(File.stat!(directory).mode, 0o777) == 0o700
+        assert_raise MatchError, fn -> S.private_file(path, "replacement") end
+        assert S.json(path) == %{"secret" => "value"}
+        directory
+      end)
+
+    refute File.exists?(directory)
+
+    assert_raise RuntimeError, "fixture failed", fn ->
+      S.temporary(fn path ->
+        send(self(), {:temporary, path})
+        raise "fixture failed"
+      end)
+    end
+
+    assert_received {:temporary, path}
+    refute File.exists?(path)
+  end
+
+  test "existing staging JSON dotenv values remain readable" do
+    S.temporary(fn directory ->
+      settings = %{
+        "SYNIXIR_STAGE_DIR" => directory,
+        "SYNIXIR_STAGE_PROJECT" => "synixir-staging",
+        "SYNIXIR_STAGE_PORT" => "8443",
+        "SYNIXIR_IMAGE" => "synixir:staging",
+        "EXTRA" => "a=b \"quoted\""
+      }
+
+      S.private_file(
+        Path.join(directory, ".env"),
+        Enum.map(settings, fn {key, value} ->
+          [key, "=", JSON.encode!(value), "\n"]
+        end)
+      )
+
+      assert Stage.open(directory).settings == settings
+      assert Stage.open(directory).url == "https://localhost:8443"
+    end)
+  end
+
+  test "restore rejects ordinary databases and corrupt archives before running PostgreSQL" do
+    assert_raise ArgumentError, fn -> Database.restore(nil, "/missing.dump", "synixir_dev") end
+
+    S.temporary(fn directory ->
+      archive = Path.join(directory, "backup.dump")
+      S.private_file(archive, "original")
+      S.private_json(archive <> ".json", %{sha256: Database.digest(archive)})
+      File.write!(archive, "corrupted")
+
+      assert_raise RuntimeError, "Archive checksum mismatch", fn ->
+        Database.restore(nil, archive, "synixir_restore_012345abcdef")
+      end
+    end)
+  end
+end
diff --git a/scripts/staging-pilot.js b/scripts/staging-pilot.js
index 862f7f1..01f41dd 100644
--- a/scripts/staging-pilot.js
+++ b/scripts/staging-pilot.js
@@ -10,11 +10,22 @@ assert.equal(new URL(baseURL).protocol, "https:");
 const browser = await chromium.launch({ headless: true });
 const contexts = [];
 const pageErrors = [];
+const failedRequests = [];
 const password = randomBytes(24).toString("hex");
 
 async function context(storageState) {
   const value = await browser.newContext({ baseURL, ignoreHTTPSErrors: true, storageState });
-  value.on("page", page => page.on("pageerror", error => pageErrors.push(error.message)));
+  value.on("page", page => {
+    page.on("pageerror", error => pageErrors.push(error.message));
+    page.on("requestfailed", request => failedRequests.push({
+      path: new URL(request.url()).pathname, error: request.failure()?.errorText,
+    }));
+    page.on("response", response => {
+      if (response.status() >= 400) failedRequests.push({
+        path: new URL(response.url()).pathname, status: response.status(),
+      });
+    });
+  });
   contexts.push(value);
   return value;
 }
@@ -42,7 +53,15 @@ async function open(ctx, roomId, role) {
   const response = await page.goto(`/?room=${roomId}`);
   assert.equal(response.status(), 200);
   assert.equal(response.headers()["cache-control"], "no-store");
-  await expect(page.locator("#status")).toHaveText("Connected", { timeout: 15000 });
+  try {
+    await expect(page.locator("#status")).toHaveText("Connected", { timeout: 15000 });
+  } catch (error) {
+    console.error("Pilot connection diagnostics:", JSON.stringify({
+      role, pageErrors, failedRequests: failedRequests.slice(-20),
+      accountError: await page.locator("#account-error").textContent(),
+    }));
+    throw error;
+  }
   await expect(page.locator("#role-label")).toHaveText(role);
   return page;
 }
diff --git a/scripts/staging.exs b/scripts/staging.exs
new file mode 100644
index 0000000..2fa186a
--- /dev/null
+++ b/scripts/staging.exs
@@ -0,0 +1,80 @@
+#!/usr/bin/env elixir
+Code.require_file("support/staging.exs", __DIR__)
+
+defmodule Synixir.Script.StagingCLI do
+  alias Synixir.Script, as: S
+  alias Synixir.Script.Stage
+
+  @commands ~w(init build up status stop backup pilot check-pilot native-check outage-check verify upgrade rehearse)
+  @usage """
+  Usage: elixir scripts/staging.exs [--directory PATH] COMMAND
+    init [--project synixir-staging] [--port 8443] [--image synixir:staging]
+    build | up | status | stop | backup | pilot | check-pilot | native-check | outage-check
+    verify ARCHIVE
+    upgrade --image IMAGE
+    rehearse --image IMAGE --output PATH [--port 8444]
+  """
+
+  def main(argv) do
+    case S.parse(
+           argv,
+           [
+             directory: :string,
+             project: :string,
+             port: :integer,
+             image: :string,
+             output: :string
+           ],
+           @commands,
+           @usage
+         ) do
+      :help -> :ok
+      {options, args} -> execute(options, args)
+    end
+  end
+
+  defp execute(options, args) do
+    directory = Path.expand(options[:directory] || Path.join(S.root(), ".local/staging"))
+
+    case args do
+      ["init"] ->
+        Stage.initialize(
+          directory,
+          options[:project] || "synixir-staging",
+          options[:port] || 8443,
+          options[:image] || "synixir:staging"
+        )
+
+      ["rehearse"] ->
+        Stage.rehearse(
+          S.required(options, :image),
+          options[:port] || 8444,
+          Path.expand(S.required(options, :output))
+        )
+
+      _ ->
+        dispatch(Stage.open(directory), args, options)
+    end
+  end
+
+  defp dispatch(stage, args, options) do
+    case args do
+      ["build"] -> Stage.build(stage)
+      ["up"] -> Stage.up(stage)
+      ["status"] -> Stage.compose(stage, ["ps"], capture: false)
+      ["stop"] -> Stage.compose(stage, ["stop"], capture: false)
+      ["backup"] -> Stage.backup(stage)
+      ["pilot"] -> Stage.pilot(stage)
+      ["check-pilot"] -> Stage.check_pilot(stage)
+      ["native-check"] -> Stage.native_check(stage)
+      ["outage-check"] -> Stage.outage_check(stage)
+      ["verify", archive] -> Stage.verify(stage, Path.expand(archive))
+      ["upgrade"] -> Stage.upgrade(stage, S.required(options, :image))
+      _ -> raise ArgumentError, @usage
+    end
+
+    if hd(args) in ~w(up status pilot), do: IO.puts("Staging: #{stage.url}")
+  end
+end
+
+Synixir.Script.cli(fn -> Synixir.Script.StagingCLI.main(System.argv()) end)
diff --git a/scripts/staging.py b/scripts/staging.py
deleted file mode 100644
index fe23348..0000000
--- a/scripts/staging.py
+++ /dev/null
@@ -1,298 +0,0 @@
-#!/usr/bin/env python3
-"""Manage an isolated, loopback-only Docker staging installation of the production release."""
-import argparse
-import json
-import os
-from pathlib import Path
-import re
-import secrets
-import ssl
-import subprocess
-import tempfile
-import time
-from urllib.request import urlopen
-from urllib.error import HTTPError
-from operations.database import Database, private_json
-
-ROOT = Path(__file__).resolve().parents[1]
-
-
-def command(args, *, capture=False, env=None, timeout=300):
-    result = subprocess.run(args, cwd=ROOT, env=env, timeout=timeout,
-                            stdout=subprocess.PIPE if capture else None,
-                            stderr=subprocess.PIPE if capture else None)
-    if result.returncode:
-        if capture:
-            # Release logs contain no generated credentials. Never print command arguments or environment.
-            print(result.stdout.decode(errors="replace")[-4000:])
-            print(result.stderr.decode(errors="replace")[-4000:])
-        raise RuntimeError(f"{args[0]} exited with {result.returncode}")
-    return result.stdout.decode() if capture else None
-
-
-def write_private(path, text):
-    with open(os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w") as out:
-        out.write(text)
-
-
-def initialize(directory, project, port, image):
-    if not re.fullmatch(r"synixir-(staging|pilot-[a-f0-9]{12})", project):
-        raise ValueError("Use synixir-staging or synixir-pilot-<12 lowercase hex digits>")
-    if not 1024 <= port <= 65535 or port in [4000, 4010, 4012, 5173, 5174, 5432]:
-        raise ValueError("Choose a separate unprivileged HTTPS port")
-    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/:@-]+", image):
-        raise ValueError("Use a Docker image tag or digest")
-    # Refuse an existing project or directory; initialization never rotates a live DB password.
-    existing = command(["docker", "ps", "-aq", "--filter", f"label=com.docker.compose.project={project}"], capture=True)
-    volumes = command(["docker", "volume", "ls", "-q", "--filter", f"label=com.docker.compose.project={project}"], capture=True)
-    if existing.strip() or volumes.strip():
-        raise ValueError("That Compose project already owns resources; retain its existing staging configuration")
-    directory.mkdir(mode=0o700, parents=True, exist_ok=False)
-    values = {"SYNIXIR_STAGE_PROJECT": project, "SYNIXIR_STAGE_PORT": str(port),
-              "SYNIXIR_STAGE_UID": str(os.getuid()), "SYNIXIR_STAGE_GID": str(os.getgid()),
-              "SYNIXIR_STAGE_DIR": str(directory), "SYNIXIR_IMAGE": image,
-              "POSTGRES_PASSWORD": secrets.token_hex(32), "SYNIXIR_DATABASE_PASSWORD": secrets.token_hex(32),
-              "SECRET_KEY_BASE": secrets.token_hex(64), "RELEASE_COOKIE": secrets.token_hex(32),
-              "SYNIXIR_METRICS_TOKEN": secrets.token_hex(32)}
-    # JSON quoting is valid for Compose dotenv values, including paths containing spaces.
-    write_private(directory / ".env", "".join(f"{key}={json.dumps(value)}\n" for key, value in values.items()))
-    command(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-sha256", "-days", "30",
-             "-keyout", str(directory / "localhost.key"), "-out", str(directory / "localhost.crt"),
-             "-subj", "/CN=localhost", "-addext", "subjectAltName=DNS:localhost,IP:127.0.0.1"], capture=True)
-    (directory / "localhost.key").chmod(0o600)
-    write_private(directory / "prometheus.yml", """global:
-  scrape_interval: 5s
-  evaluation_interval: 5s
-rule_files:
-  - /etc/prometheus/alerts.yml
-scrape_configs:
-  - job_name: synixir
-    scrape_timeout: 3s
-    authorization:
-      credentials: """ + json.dumps(values["SYNIXIR_METRICS_TOKEN"]) + """
-    static_configs:
-      - targets: ["app:4000"]
-""")
-    print(f"Initialized {project} in {directory}. Credentials and TLS key are private files.")
-
-
-class Stage:
-    def __init__(self, directory):
-        self.directory = directory
-        self.settings = dict(line.split("=", 1) for line in (directory / ".env").read_text().splitlines() if line)
-        self.settings = {key: json.loads(value) for key, value in self.settings.items()}
-        if Path(self.settings["SYNIXIR_STAGE_DIR"]) != directory:
-            raise ValueError("Staging configuration belongs to a different directory")
-        if not re.fullmatch(r"synixir-(staging|pilot-[a-f0-9]{12})", self.settings["SYNIXIR_STAGE_PROJECT"]):
-            raise ValueError("Invalid staging project")
-        self.url = "https://localhost:" + self.settings["SYNIXIR_STAGE_PORT"]
-
-    def compose(self, *args, capture=False, overrides=None, timeout=300):
-        # Explicit values prevent an unrelated shell variable from selecting another project or volume.
-        env = dict(os.environ, **self.settings)
-        env.update(overrides or {})
-        return command(["docker", "compose", "--env-file", str(self.directory / ".env"),
-                        "-p", self.settings["SYNIXIR_STAGE_PROJECT"], "-f", "deploy/compose.yaml", *args],
-                       capture=capture, env=env, timeout=timeout)
-
-    def build(self):
-        revision = command(["git", "rev-parse", "HEAD"], capture=True).strip()
-        if command(["git", "status", "--porcelain"], capture=True).strip():
-            revision += "-dirty"
-        command(["docker", "build", "--build-arg", f"REVISION={revision}",
-                 "-t", self.settings["SYNIXIR_IMAGE"], "."], timeout=1800)
-
-    def up(self):
-        self.compose("up", "-d", "--wait", "--wait-timeout", "120")
-        self.ready()
-
-    def ready(self):
-        context = ssl.create_default_context(cafile=str(self.directory / "localhost.crt"))
-        deadline = time.monotonic() + 30
-        while True:
-            try:
-                with urlopen(self.url + "/health/ready", context=context, timeout=3) as response:
-                    if response.status == 200:
-                        return
-            except OSError:
-                if time.monotonic() >= deadline:
-                    raise RuntimeError("HTTPS staging readiness did not succeed")
-                time.sleep(0.2)
-
-    def database(self):
-        container = self.compose("ps", "-q", "db", capture=True).strip()
-        if not container:
-            raise RuntimeError("Staging database is not running")
-        return Database(container)
-
-    def native_check(self):
-        output = self.compose("run", "--rm", "--no-deps", "app", "/app/bin/synixir", "eval",
-                              "Synixir.Release.native_check()", capture=True)
-        line = next(line for line in output.splitlines() if line.startswith("NATIVE_RESULT="))
-        result = json.loads(line.split("=", 1)[1])
-        print(json.dumps(result, indent=2))
-        return result
-
-    def outage_check(self):
-        context = ssl.create_default_context(cafile=str(self.directory / "localhost.crt"))
-        try:
-            self.compose("stop", "db")
-            with urlopen(self.url + "/health/live", context=context, timeout=5) as response:
-                assert response.status == 200, "liveness must survive database outage"
-            try:
-                urlopen(self.url + "/health/ready", context=context, timeout=5)
-            except HTTPError as error:
-                assert error.code == 503, "readiness must report database outage"
-            else:
-                raise RuntimeError("Readiness accepted an unavailable database")
-        finally:
-            self.compose("start", "db")
-            self.ready()
-
-    def backup(self):
-        directory = self.directory / "backups"
-        directory.mkdir(mode=0o700, exist_ok=True)
-        archive = directory / (time.strftime("%Y%m%dT%H%M%S") + "-" + secrets.token_hex(3) + ".dump")
-        metadata = self.database().backup("synixir", archive)
-        print(f"Backup saved to {archive} ({metadata['bytes']} bytes)")
-        return archive
-
-    def verify(self, archive):
-        db = self.database()
-        with db.fresh("restore") as target:
-            db.sql("postgres", f'ALTER DATABASE "{target}" OWNER TO synixir;')
-            db.restore(archive, target, role="synixir")
-            url = f"ecto://synixir:{self.settings['SYNIXIR_DATABASE_PASSWORD']}@db/{target}"
-            output = self.compose("run", "--rm", "--no-deps", "-e", "DATABASE_URL", "app",
-                                  "/app/bin/synixir", "eval", "Synixir.Release.verify_restore()",
-                                  capture=True, overrides={"DATABASE_URL": url})
-            line = next(line for line in output.splitlines() if line.startswith("RESTORE_RESULT="))
-            result = json.loads(line.split("=", 1)[1])
-        print(json.dumps(result, indent=2))
-        return result
-
-    def upgrade(self, image):
-        target_id = command(["docker", "image", "inspect", image, "--format", "{{.Id}}"], capture=True).strip()
-        container = self.compose("ps", "-q", "app", capture=True).strip()
-        previous_id = command(["docker", "inspect", container, "--format", "{{.Image}}"], capture=True).strip()
-        archive = self.backup()
-        self.verify(archive)
-        private_json(Path(str(archive) + ".upgrade.json"), {"previous_image": previous_id,
-                     "target_image": target_id, "target_reference": image, "backup": str(archive)})
-        # One document owner per room: stop the old app before starting the replacement.
-        self.compose("stop", "gateway", "app")
-        override = {"SYNIXIR_IMAGE": image}
-        try:
-            self.compose("run", "--rm", "--no-deps", "migrate", overrides=override)
-            self.compose("up", "-d", "--no-deps", "--force-recreate", "--wait", "--wait-timeout", "120", "app", overrides=override)
-            self.compose("up", "-d", "--no-deps", "--force-recreate", "gateway", "prometheus", overrides=override)
-            self.ready()
-        except BaseException:
-            self.compose("stop", "gateway", "app")
-            raise RuntimeError("Upgrade failed; app and gateway are stopped. Preserve the backup and inspect migration logs before rollback.") from None
-        self.settings["SYNIXIR_IMAGE"] = image
-        temporary = self.directory / (".env-" + secrets.token_hex(6))
-        write_private(temporary, "".join(f"{key}={json.dumps(value)}\n" for key, value in self.settings.items()))
-        temporary.replace(self.directory / ".env")
-
-    def pilot(self):
-        manifest = self.directory / "pilot.json"
-        if manifest.exists():
-            raise ValueError("Pilot already ran in this installation; use check-pilot to verify it")
-        native = self.native_check()
-        private_json(manifest, {})
-        try:
-            command(["node", "scripts/staging-pilot.js", "seed", self.url, str(manifest)])
-        except BaseException:
-            if json.loads(manifest.read_text()) == {}:
-                manifest.unlink()
-            raise
-        self.outage_check()
-        self.upgrade(self.settings["SYNIXIR_IMAGE"])
-        command(["node", "scripts/staging-pilot.js", "verify", self.url, str(manifest)])
-        restored = self.verify(self.backup())
-        deadline = time.monotonic() + 30
-        while True:
-            output = self.compose("exec", "-T", "prometheus", "wget", "-qO-",
-                                  "http://localhost:9090/api/v1/targets", capture=True)
-            targets = json.loads(output)["data"]["activeTargets"]
-            if targets and all(target["health"] == "up" for target in targets):
-                break
-            if time.monotonic() >= deadline:
-                raise RuntimeError("Prometheus did not scrape the private metrics endpoint")
-            time.sleep(1)
-        report = {"checked_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
-                  "image": self.settings["SYNIXIR_IMAGE"], "https": True, "secure_cookies": True,
-                  "collaboration_and_roles": True, "offline_reconnect": True, "same_image_upgrade": True,
-                  "private_metrics_scraped": True, "restored_database": restored,
-                  "native_libraries": native, "database_outage_recovery": True}
-        report["image_id"] = command(["docker", "image", "inspect", self.settings["SYNIXIR_IMAGE"],
-                                      "--format", "{{.Id}}"], capture=True).strip()
-        private_json(self.directory / "pilot-result.json", report)
-        print(json.dumps(report, indent=2))
-
-
-def rehearse(image, port, output):
-    if output.exists():
-        raise ValueError("Rehearsal output exists; choose a new path")
-    local = ROOT / ".local"
-    local.mkdir(exist_ok=True)
-    with tempfile.TemporaryDirectory(prefix="release-pilot-", dir=local) as temporary:
-        directory = Path(temporary) / "stage"
-        project = "synixir-pilot-" + secrets.token_hex(6)
-        initialize(directory, project, port, image)
-        stage = Stage(directory)
-        try:
-            stage.up()
-            stage.pilot()
-            private_json(output, json.loads((directory / "pilot-result.json").read_text()))
-        finally:
-            # Only remove the new project and fixture volume owned by this invocation.
-            stage.compose("down", "--volumes")
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--directory", type=Path, default=ROOT / ".local/staging")
-    commands = parser.add_subparsers(dest="command", required=True)
-    init = commands.add_parser("init")
-    init.add_argument("--project", default="synixir-staging")
-    init.add_argument("--port", type=int, default=8443)
-    init.add_argument("--image", default="synixir:staging")
-    for name in ["build", "up", "status", "stop", "backup", "pilot", "check-pilot", "native-check", "outage-check"]:
-        commands.add_parser(name)
-    verify = commands.add_parser("verify")
-    verify.add_argument("archive", type=Path)
-    upgrade = commands.add_parser("upgrade")
-    upgrade.add_argument("--image", required=True)
-    rehearsal = commands.add_parser("rehearse")
-    rehearsal.add_argument("--image", required=True)
-    rehearsal.add_argument("--port", type=int, default=8444)
-    rehearsal.add_argument("--output", type=Path, required=True)
-    args = parser.parse_args()
-    directory = args.directory.resolve()
-    if args.command == "rehearse":
-        rehearse(args.image, args.port, args.output.resolve())
-        return
-    if args.command == "init":
-        initialize(directory, args.project, args.port, args.image)
-        return
-    stage = Stage(directory)
-    if args.command == "verify":
-        stage.verify(args.archive.resolve())
-    elif args.command == "upgrade":
-        stage.upgrade(args.image)
-    elif args.command == "status":
-        stage.compose("ps")
-    elif args.command == "stop":
-        stage.compose("stop")
-    elif args.command == "check-pilot":
-        command(["node", "scripts/staging-pilot.js", "verify", stage.url, str(directory / "pilot.json")])
-    else:
-        getattr(stage, args.command.replace("-", "_"))()
-    if args.command in ["up", "status", "pilot"]:
-        print(f"Staging: {stage.url}")
-
-
-if __name__ == "__main__":
-    main()
diff --git a/scripts/support/database.exs b/scripts/support/database.exs
new file mode 100644
index 0000000..f74e134
--- /dev/null
+++ b/scripts/support/database.exs
@@ -0,0 +1,227 @@
+Code.require_file("operations.exs", __DIR__)
+
+defmodule Synixir.Script.Database do
+  @moduledoc "PostgreSQL backups and isolated restore/load drills."
+  alias Synixir.Script, as: S
+  @isolated ~r/\Asynixir_(load|drill|restore)_[a-f0-9]{12}\z/
+
+  def command(container, tool, args) do
+    prefix = if container, do: ["docker", "exec", "-i", container], else: []
+
+    user =
+      if container && "--version" not in args,
+        do: ["--username", System.get_env("PGUSER", "postgres")],
+        else: []
+
+    prefix ++ [tool] ++ user ++ args
+  end
+
+  def sql(container, database, query) do
+    command(container, "psql", [
+      "--no-psqlrc",
+      "--set",
+      "ON_ERROR_STOP=1",
+      "--dbname",
+      database,
+      "--tuples-only",
+      "--no-align",
+      "--command",
+      query
+    ])
+    |> S.run()
+    |> String.trim()
+  end
+
+  def isolated!(name) do
+    if not Regex.match?(@isolated, name),
+      do: raise(ArgumentError, "Expected an isolated operations database")
+
+    name
+  end
+
+  def fresh(container, kind, fun) do
+    name = isolated!("synixir_#{kind}_#{S.token()}")
+    sql(container, "postgres", ~s(CREATE DATABASE "#{name}" TEMPLATE template0;))
+
+    try do
+      fun.(name)
+    after
+      sql(container, "postgres", ~s(DROP DATABASE "#{name}" WITH \(FORCE\);))
+    end
+  end
+
+  def digest(path) do
+    path
+    |> File.stream!([], 64 * 1024)
+    |> Enum.reduce(:crypto.hash_init(:sha256), &:crypto.hash_update(&2, &1))
+    |> :crypto.hash_final()
+    |> Base.encode16(case: :lower)
+  end
+
+  def backup(container, database, archive) do
+    if not Regex.match?(~r/\A[A-Za-z_][A-Za-z0-9_]{0,62}\z/, database),
+      do: raise(ArgumentError, "Pass a plain database name, not a connection string")
+
+    S.unused!(archive)
+    S.unused!(archive <> ".json")
+    started = S.now()
+
+    migrations =
+      sql(container, database, "SELECT version FROM schema_migrations ORDER BY version")
+      |> String.split("\n", trim: true)
+
+    version = command(container, "pg_dump", ["--version"]) |> S.run() |> String.trim()
+    S.private_file(archive, "")
+
+    try do
+      command(container, "pg_dump", ["--format=custom", "--dbname", database])
+      |> S.run(stdout: archive)
+    rescue
+      error ->
+        File.rm!(archive)
+        reraise error, __STACKTRACE__
+    end
+
+    metadata = %{
+      "started_at" => started,
+      "finished_at" => S.now(),
+      "database" => database,
+      "tool" => version,
+      "migrations_before_dump" => migrations,
+      "bytes" => File.stat!(archive).size,
+      "sha256" => digest(archive)
+    }
+
+    S.private_json(archive <> ".json", metadata)
+    metadata
+  end
+
+  def restore(container, archive, target, role \\ nil) do
+    isolated!(target)
+
+    if digest(archive) != S.json(archive <> ".json")["sha256"],
+      do: raise("Archive checksum mismatch")
+
+    args = [
+      "--single-transaction",
+      "--exit-on-error",
+      "--no-owner",
+      "--no-privileges",
+      "--dbname",
+      target
+    ]
+
+    command(container, "pg_restore", args ++ if(role, do: ["--role", role], else: []))
+    |> S.run(stdin: archive)
+  end
+
+  def app_env(database) do
+    isolated!(database)
+
+    [{"MIX_ENV", "test"}, {"SYNIXIR_OPERATIONS_DATABASE", database}] ++
+      Enum.map(~w(PHX_SERVER SYNIXIR_BROWSER_TEST DATABASE_URL MIX_TEST_PARTITION), &{&1, nil})
+  end
+
+  def mix(database, args) do
+    output = S.run(["mix" | args], env: app_env(database))
+    if String.contains?(output, "OPERATIONS_RESULT="), do: S.result(output, "OPERATIONS_RESULT=")
+  end
+
+  def check(database, args), do: mix(database, ["run", "scripts/operations/restore.exs" | args])
+
+  def verify(container, archive, fixture \\ nil) do
+    started = S.monotonic()
+
+    result =
+      fresh(container, "restore", fn target ->
+        restore(container, archive, target)
+        check(target, if(fixture, do: ["drill-verify", fixture], else: ["verify"]))
+      end)
+
+    Map.put(result, "restore_and_check_seconds", Float.round((S.monotonic() - started) / 1000, 3))
+  end
+
+  def drill(container) do
+    S.temporary(fn directory ->
+      fresh(container, "drill", fn source ->
+        mix(source, ["ecto.migrate", "--quiet"])
+        mix(source, ["run", "scripts/operations/limits.exs"])
+        fixture = Path.join(directory, "expectations.json")
+        S.private_json(fixture, %{})
+        check(source, ["seed", fixture])
+        archive = Path.join(directory, "backup.dump")
+        backup = backup(container, source, archive)
+
+        Map.merge(verify(container, archive, fixture), %{
+          "archive_bytes" => backup["bytes"],
+          "recorded_at" => backup["finished_at"],
+          "backup_tool" => backup["tool"],
+          "isolated_limits_checks" => true
+        })
+      end)
+    end)
+  end
+
+  def load(container, clients, rooms, writes, port, output) do
+    unless clients in 1..100 and rooms in 1..clients and writes in 1..1000,
+      do: raise(ArgumentError, "Use 1-100 clients, 1-clients rooms, and 1-1000 writes per client")
+
+    unless port in 1024..65535 and port not in [4000, 4010, 5173, 5174],
+      do:
+        raise(
+          ArgumentError,
+          "Choose an unused unprivileged port separate from development/browser fixtures"
+        )
+
+    {:ok, probe} = :gen_tcp.listen(port, ip: {127, 0, 0, 1})
+    :ok = :gen_tcp.close(probe)
+
+    fresh(container, "load", fn database ->
+      mix(database, ["ecto.migrate", "--quiet"])
+
+      env =
+        Map.new(app_env(database))
+        |> Map.merge(%{
+          "PHX_SERVER" => "true",
+          "PORT" => to_string(port),
+          "SYNIXIR_METRICS_TOKEN" => S.token(32)
+        })
+
+      child = S.start(["mix", "phx.server"], env: env)
+
+      {metrics, report} =
+        try do
+          S.eventually(
+            fn ->
+              if not S.running?(child), do: raise("Isolated load server exited")
+              S.http_status("http://127.0.0.1:#{port}/health/ready", timeout: 1000) == 200
+            end,
+            "Isolated load server did not start"
+          )
+
+          S.run(
+            [
+              "node",
+              "scripts/operations/load.js"
+              | Enum.map([port, clients, rooms, writes], &to_string/1)
+            ],
+            env: env
+          )
+          |> JSON.decode!()
+          |> Map.pop!("metrics_text")
+        rescue
+          error ->
+            S.log_tail(child.stdout)
+            S.log_tail(child.stderr)
+            reraise error, __STACKTRACE__
+        after
+          S.stop(child)
+        end
+
+      report = Map.put(report, "stored_documents", check(database, ["verify"]))
+      S.private_file(output <> ".prom", metrics)
+      S.private_json(output, report)
+      report
+    end)
+  end
+end
diff --git a/scripts/support/operations.exs b/scripts/support/operations.exs
new file mode 100644
index 0000000..4c5cb74
--- /dev/null
+++ b/scripts/support/operations.exs
@@ -0,0 +1,278 @@
+defmodule Synixir.Script do
+  @moduledoc "Shared, dependency-free support for the operations CLIs."
+  @root Path.expand("../..", __DIR__)
+
+  def root, do: @root
+  def token(bytes \\ 6), do: Base.encode16(:crypto.strong_rand_bytes(bytes), case: :lower)
+  def now, do: DateTime.utc_now() |> DateTime.truncate(:second) |> DateTime.to_iso8601()
+  def monotonic, do: System.monotonic_time(:millisecond)
+  def json(path), do: path |> File.read!() |> JSON.decode!()
+  def print(value), do: IO.puts(JSON.encode!(value))
+
+  def private_file(path, content) do
+    {:ok, file} = :file.open(String.to_charlist(path), [:write, :binary, :exclusive])
+
+    try do
+      File.chmod!(path, 0o600)
+      :ok = :file.write(file, content)
+    after
+      :file.close(file)
+    end
+  end
+
+  def private_json(path, value), do: private_file(path, [JSON.encode!(value), "\n"])
+
+  def private_directory(path) do
+    File.mkdir!(path)
+    File.chmod!(path, 0o700)
+    path
+  end
+
+  def temporary(fun, parent \\ System.tmp_dir!()) do
+    directory = private_directory(Path.join(parent, "synixir-" <> token()))
+
+    try do
+      fun.(directory)
+    after
+      File.rm_rf!(directory)
+    end
+  end
+
+  # Redirect binary archives through files, never through a shell-interpolated command
+  # or an in-memory string. Every command argument is passed as a separate argv value.
+  def start([program | args], options \\ []) do
+    executable = System.find_executable(program) || raise "Required command not found: #{program}"
+    directory = private_directory(Path.join(System.tmp_dir!(), "synixir-command-" <> token()))
+    stdout = Keyword.get(options, :stdout, Path.join(directory, "stdout"))
+    stderr = Path.join(directory, "stderr")
+    stdin = Keyword.get(options, :stdin, "/dev/null")
+    if not Keyword.has_key?(options, :stdout), do: private_file(stdout, "")
+    private_file(stderr, "")
+
+    env =
+      Enum.map(Keyword.get(options, :env, []), fn {key, value} ->
+        {String.to_charlist(key), if(value, do: String.to_charlist(value), else: false)}
+      end)
+
+    try do
+      port =
+        Port.open({:spawn_executable, System.find_executable("sh")}, [
+          :binary,
+          :exit_status,
+          :hide,
+          args: [
+            "-c",
+            "input=$1; output=$2; errors=$3; shift 3; exec \"$@\" <\"$input\" >\"$output\" 2>\"$errors\"",
+            "synixir-command",
+            stdin,
+            stdout,
+            stderr,
+            executable | args
+          ],
+          cd: @root,
+          env: env
+        ])
+
+      {:os_pid, pid} = Port.info(port, :os_pid)
+
+      %{
+        port: port,
+        pid: pid,
+        directory: directory,
+        stdout: stdout,
+        stderr: stderr,
+        program: program
+      }
+    rescue
+      error ->
+        File.rm_rf!(directory)
+        reraise error, __STACKTRACE__
+    end
+  end
+
+  def run(args, options \\ []) do
+    child = start(args, options)
+
+    try do
+      receive do
+        {port, {:exit_status, status}} when port == child.port ->
+          unless status == 0 do
+            log_tail(child.stderr)
+            unless Keyword.has_key?(options, :stdout), do: log_tail(child.stdout)
+            raise "#{child.program} exited with #{status}"
+          end
+
+          unless Keyword.has_key?(options, :stdout) do
+            output = File.read!(child.stdout)
+            if Keyword.get(options, :capture, true), do: output, else: IO.write(output)
+          end
+      after
+        Keyword.get(options, :timeout, 300_000) ->
+          raise "#{child.program} timed out"
+      end
+    after
+      stop(child)
+    end
+  end
+
+  def running?(child), do: Port.info(child.port) != nil
+
+  def stop(child) do
+    if running?(child) do
+      # Mix can launch a BEAM child. Stop descendants as well before removing fixtures.
+      {processes, 0} = System.cmd("ps", ["-axo", "pid=,ppid="])
+      pairs = Enum.map(String.split(processes, "\n", trim: true), &String.split/1)
+      pids = descendants(pairs, Integer.to_string(child.pid)) ++ [Integer.to_string(child.pid)]
+      System.cmd("kill", ["-TERM" | pids], stderr_to_stdout: true)
+      deadline = monotonic() + 10_000
+      await_exit(child, deadline)
+      # Descendants may outlive the port's immediate child.
+      alive =
+        Enum.filter(pids, fn pid ->
+          elem(System.cmd("kill", ["-0", pid], stderr_to_stdout: true), 1) == 0
+        end)
+
+      if alive != [], do: System.cmd("kill", ["-KILL" | alive], stderr_to_stdout: true)
+      if running?(child), do: Port.close(child.port)
+    end
+
+    File.rm_rf!(child.directory)
+    :ok
+  end
+
+  defp descendants(pairs, parent) do
+    for [pid, ppid] <- pairs, ppid == parent, child <- descendants(pairs, pid) ++ [pid], do: child
+  end
+
+  defp await_exit(child, deadline) do
+    if running?(child) and monotonic() < deadline do
+      Process.sleep(50)
+      await_exit(child, deadline)
+    end
+  end
+
+  def log_tail(path) do
+    {:ok, file} = :file.open(String.to_charlist(path), [:read, :binary])
+
+    try do
+      {:ok, size} = :file.position(file, :eof)
+      {:ok, _} = :file.position(file, max(0, size - 6000))
+
+      case :file.read(file, 6000) do
+        {:ok, content} -> IO.binwrite(:stderr, content)
+        :eof -> :ok
+      end
+    after
+      :file.close(file)
+    end
+  end
+
+  def http_status(url, options \\ []) do
+    {:ok, _} = Application.ensure_all_started(:inets)
+    {:ok, _} = Application.ensure_all_started(:ssl)
+
+    ssl =
+      case options[:cacert] do
+        nil ->
+          []
+
+        path ->
+          certificates =
+            for {:Certificate, der, :not_encrypted} <- :public_key.pem_decode(File.read!(path)),
+                do: der
+
+          [
+            verify: :verify_peer,
+            cacertfile: String.to_charlist(path),
+            verify_fun: {&verify_certificate/3, {certificates, URI.parse(url).host}},
+            log_level: :error,
+            customize_hostname_check: [
+              match_fun: :public_key.pkix_verify_hostname_match_fun(:https)
+            ]
+          ]
+      end
+
+    case :httpc.request(
+           :get,
+           {String.to_charlist(url), []},
+           [
+             timeout: options[:timeout] || 3000,
+             connect_timeout: options[:timeout] || 3000,
+             autoredirect: false,
+             ssl: ssl
+           ],
+           []
+         ) do
+      {:ok, {{_, status, _}, _, _}} -> status
+      {:error, _} -> nil
+    end
+  end
+
+  # OTP reports a self-signed leaf separately even when it is in cacertfile.
+  # Trust only the saved certificate, and still validate its dates and hostname.
+  defp verify_certificate(certificate, {:bad_cert, :selfsigned_peer}, {trusted, host} = state) do
+    der = :public_key.pkix_encode(:OTPCertificate, certificate, :otp)
+
+    valid =
+      der in trusted and
+        match?({:ok, _}, :public_key.pkix_path_validation(certificate, [der], [])) and
+        :public_key.pkix_verify_hostname(certificate, [{:dns_id, String.to_charlist(host)}],
+          match_fun: :public_key.pkix_verify_hostname_match_fun(:https)
+        )
+
+    if valid, do: {:valid, state}, else: {:fail, :untrusted_staging_certificate}
+  end
+
+  defp verify_certificate(_, {:bad_cert, reason}, _), do: {:fail, reason}
+  defp verify_certificate(_, {:extension, _}, state), do: {:unknown, state}
+
+  defp verify_certificate(_, event, state) when event in [:valid, :valid_peer],
+    do: {:valid, state}
+
+  def eventually(fun, message, timeout \\ 30_000) do
+    poll(fun, message, monotonic() + timeout)
+  end
+
+  defp poll(fun, message, deadline) do
+    unless fun.() do
+      if monotonic() >= deadline, do: raise(message)
+      Process.sleep(200)
+      poll(fun, message, deadline)
+    end
+  end
+
+  def result(output, prefix) do
+    line = Enum.find(String.split(output, "\n"), &String.starts_with?(&1, prefix))
+    if is_nil(line), do: raise("Command did not return #{prefix}")
+    line |> String.replace_prefix(prefix, "") |> JSON.decode!()
+  end
+
+  def parse(argv, switches, commands, usage) do
+    {options, args, invalid} = OptionParser.parse(argv, strict: [help: :boolean] ++ switches)
+
+    if options[:help] do
+      IO.puts(usage)
+      :help
+    else
+      if invalid != [] or args == [] or hd(args) not in commands, do: raise(ArgumentError, usage)
+      {options, args}
+    end
+  end
+
+  def required(options, key), do: options[key] || raise(ArgumentError, "Missing --#{key}")
+
+  def unused!(path) do
+    if File.exists?(path), do: raise(ArgumentError, "Output exists; choose a new path: #{path}")
+    path
+  end
+
+  def cli(fun) do
+    try do
+      fun.()
+    rescue
+      error ->
+        IO.puts(:stderr, Exception.message(error))
+        System.halt(1)
+    end
+  end
+end
diff --git a/scripts/support/staging.exs b/scripts/support/staging.exs
new file mode 100644
index 0000000..e4a2416
--- /dev/null
+++ b/scripts/support/staging.exs
@@ -0,0 +1,420 @@
+Code.require_file("database.exs", __DIR__)
+
+defmodule Synixir.Script.Stage do
+  @moduledoc "Manage an isolated, loopback-only Docker staging installation."
+  alias Synixir.Script, as: S
+  alias Synixir.Script.Database
+  @project ~r/\Asynixir-(staging|pilot-[a-f0-9]{12})\z/
+
+  def initialize(directory, project, port, image) do
+    unless Regex.match?(@project, project),
+      do: raise(ArgumentError, "Use synixir-staging or synixir-pilot-<12 lowercase hex digits>")
+
+    unless port in 1024..65535 and port not in [4000, 4010, 4012, 5173, 5174, 5432],
+      do: raise(ArgumentError, "Choose a separate unprivileged HTTPS port")
+
+    unless Regex.match?(~r/\A[A-Za-z0-9][A-Za-z0-9._\/:@-]+\z/, image),
+      do: raise(ArgumentError, "Use a Docker image tag or digest")
+
+    existing =
+      S.run(["docker", "ps", "-aq", "--filter", "label=com.docker.compose.project=#{project}"])
+
+    volumes =
+      S.run([
+        "docker",
+        "volume",
+        "ls",
+        "-q",
+        "--filter",
+        "label=com.docker.compose.project=#{project}"
+      ])
+
+    unless String.trim(existing) == "" and String.trim(volumes) == "",
+      do:
+        raise(
+          "That Compose project already owns resources; retain its existing staging configuration"
+        )
+
+    File.mkdir_p!(Path.dirname(directory))
+    S.private_directory(directory)
+
+    values = %{
+      "SYNIXIR_STAGE_PROJECT" => project,
+      "SYNIXIR_STAGE_PORT" => to_string(port),
+      "SYNIXIR_STAGE_UID" => S.run(["id", "-u"]) |> String.trim(),
+      "SYNIXIR_STAGE_GID" => S.run(["id", "-g"]) |> String.trim(),
+      "SYNIXIR_STAGE_DIR" => directory,
+      "SYNIXIR_IMAGE" => image,
+      "POSTGRES_PASSWORD" => S.token(32),
+      "SYNIXIR_DATABASE_PASSWORD" => S.token(32),
+      "SECRET_KEY_BASE" => S.token(64),
+      "RELEASE_COOKIE" => S.token(32),
+      "SYNIXIR_METRICS_TOKEN" => S.token(32)
+    }
+
+    S.private_file(Path.join(directory, ".env"), dotenv(values))
+
+    S.run([
+      "openssl",
+      "req",
+      "-x509",
+      "-newkey",
+      "rsa:2048",
+      "-nodes",
+      "-sha256",
+      "-days",
+      "30",
+      "-keyout",
+      Path.join(directory, "localhost.key"),
+      "-out",
+      Path.join(directory, "localhost.crt"),
+      "-subj",
+      "/CN=localhost",
+      "-addext",
+      "subjectAltName=DNS:localhost,IP:127.0.0.1"
+    ])
+
+    File.chmod!(Path.join(directory, "localhost.key"), 0o600)
+
+    S.private_file(Path.join(directory, "prometheus.yml"), """
+    global:
+      scrape_interval: 5s
+      evaluation_interval: 5s
+    rule_files:
+      - /etc/prometheus/alerts.yml
+    scrape_configs:
+      - job_name: synixir
+        scrape_timeout: 3s
+        authorization:
+          credentials: #{JSON.encode!(values["SYNIXIR_METRICS_TOKEN"])}
+        static_configs:
+          - targets: ["app:4000"]
+    """)
+
+    IO.puts("Initialized #{project} in #{directory}. Credentials and TLS key are private files.")
+  end
+
+  defp dotenv(settings),
+    do: Enum.map(settings, fn {key, value} -> [key, "=", JSON.encode!(value), "\n"] end)
+
+  def open(directory) do
+    settings =
+      Path.join(directory, ".env")
+      |> File.read!()
+      |> String.split("\n", trim: true)
+      |> Map.new(fn line ->
+        [key, value] = String.split(line, "=", parts: 2)
+        {key, JSON.decode!(value)}
+      end)
+
+    unless Path.expand(settings["SYNIXIR_STAGE_DIR"]) == directory,
+      do: raise("Staging configuration belongs to a different directory")
+
+    unless Regex.match?(@project, settings["SYNIXIR_STAGE_PROJECT"]),
+      do: raise("Invalid staging project")
+
+    %{
+      directory: directory,
+      settings: settings,
+      url: "https://localhost:" <> settings["SYNIXIR_STAGE_PORT"]
+    }
+  end
+
+  def compose(stage, args, options \\ []) do
+    env = Map.merge(stage.settings, Keyword.get(options, :overrides, %{}))
+
+    S.run(
+      [
+        "docker",
+        "compose",
+        "--env-file",
+        Path.join(stage.directory, ".env"),
+        "-p",
+        stage.settings["SYNIXIR_STAGE_PROJECT"],
+        "-f",
+        "deploy/compose.yaml" | args
+      ],
+      options |> Keyword.delete(:overrides) |> Keyword.put(:env, env)
+    )
+  end
+
+  def build(stage) do
+    revision = S.run(["git", "rev-parse", "HEAD"]) |> String.trim()
+
+    revision =
+      if String.trim(S.run(["git", "status", "--porcelain"])) == "",
+        do: revision,
+        else: revision <> "-dirty"
+
+    S.run(
+      [
+        "docker",
+        "build",
+        "--build-arg",
+        "REVISION=#{revision}",
+        "-t",
+        stage.settings["SYNIXIR_IMAGE"],
+        "."
+      ],
+      timeout: 1_800_000,
+      capture: false
+    )
+  end
+
+  def up(stage) do
+    compose(stage, ["up", "-d", "--wait", "--wait-timeout", "120"], capture: false)
+    ready(stage)
+  end
+
+  def ready(stage) do
+    S.eventually(
+      fn -> health(stage, "ready") == 200 end,
+      "HTTPS staging readiness did not succeed"
+    )
+  end
+
+  defp health(stage, probe),
+    do:
+      S.http_status(stage.url <> "/health/" <> probe,
+        cacert: Path.join(stage.directory, "localhost.crt")
+      )
+
+  def database(stage) do
+    container = compose(stage, ["ps", "-q", "db"]) |> String.trim()
+    if container == "", do: raise("Staging database is not running")
+    container
+  end
+
+  def native_check(stage) do
+    result =
+      compose(stage, [
+        "run",
+        "--rm",
+        "--no-deps",
+        "app",
+        "/app/bin/synixir",
+        "eval",
+        "Synixir.Release.native_check()"
+      ])
+      |> S.result("NATIVE_RESULT=")
+
+    S.print(result)
+    result
+  end
+
+  def outage_check(stage) do
+    try do
+      compose(stage, ["stop", "db"])
+      unless health(stage, "live") == 200, do: raise("Liveness must survive database outage")
+      unless health(stage, "ready") == 503, do: raise("Readiness must report database outage")
+    after
+      compose(stage, ["start", "db"])
+      ready(stage)
+    end
+  end
+
+  def backup(stage) do
+    directory = Path.join(stage.directory, "backups")
+    unless File.dir?(directory), do: S.private_directory(directory)
+
+    archive =
+      Path.join(
+        directory,
+        Calendar.strftime(DateTime.utc_now(), "%Y%m%dT%H%M%S") <> "-" <> S.token(3) <> ".dump"
+      )
+
+    metadata = Database.backup(database(stage), "synixir", archive)
+    IO.puts("Backup saved to #{archive} (#{metadata["bytes"]} bytes)")
+    archive
+  end
+
+  def verify(stage, archive) do
+    db = database(stage)
+
+    result =
+      Database.fresh(db, "restore", fn target ->
+        Database.sql(db, "postgres", ~s(ALTER DATABASE "#{target}" OWNER TO synixir;))
+        Database.restore(db, archive, target, "synixir")
+        url = "ecto://synixir:#{stage.settings["SYNIXIR_DATABASE_PASSWORD"]}@db/#{target}"
+
+        compose(
+          stage,
+          [
+            "run",
+            "--rm",
+            "--no-deps",
+            "-e",
+            "DATABASE_URL",
+            "app",
+            "/app/bin/synixir",
+            "eval",
+            "Synixir.Release.verify_restore()"
+          ],
+          overrides: %{"DATABASE_URL" => url}
+        )
+        |> S.result("RESTORE_RESULT=")
+      end)
+
+    S.print(result)
+    result
+  end
+
+  def upgrade(stage, image) do
+    target_id =
+      S.run(["docker", "image", "inspect", image, "--format", "{{.Id}} "]) |> String.trim()
+
+    container = compose(stage, ["ps", "-q", "app"]) |> String.trim()
+
+    previous_id =
+      S.run(["docker", "inspect", container, "--format", "{{.Image}} "]) |> String.trim()
+
+    archive = backup(stage)
+    verify(stage, archive)
+
+    S.private_json(archive <> ".upgrade.json", %{
+      previous_image: previous_id,
+      target_image: target_id,
+      target_reference: image,
+      backup: archive
+    })
+
+    # One document owner per room: stop the old app before starting its replacement.
+    compose(stage, ["stop", "gateway", "app"])
+    override = [overrides: %{"SYNIXIR_IMAGE" => image}]
+
+    try do
+      compose(stage, ["run", "--rm", "--no-deps", "migrate"], override)
+
+      compose(
+        stage,
+        ["up", "-d", "--no-deps", "--force-recreate", "--wait", "--wait-timeout", "120", "app"],
+        override
+      )
+
+      compose(
+        stage,
+        ["up", "-d", "--no-deps", "--force-recreate", "gateway", "prometheus"],
+        override
+      )
+
+      ready(stage)
+    rescue
+      _ ->
+        compose(stage, ["stop", "gateway", "app"])
+
+        raise "Upgrade failed; app and gateway are stopped. Preserve the backup and inspect migration logs before rollback."
+    end
+
+    settings = Map.put(stage.settings, "SYNIXIR_IMAGE", image)
+    temporary = Path.join(stage.directory, ".env-" <> S.token())
+    S.private_file(temporary, dotenv(settings))
+    File.rename!(temporary, Path.join(stage.directory, ".env"))
+    %{stage | settings: settings}
+  end
+
+  def check_pilot(stage),
+    do:
+      S.run(
+        [
+          "node",
+          "scripts/staging-pilot.js",
+          "verify",
+          stage.url,
+          Path.join(stage.directory, "pilot.json")
+        ],
+        capture: false
+      )
+
+  def pilot(stage) do
+    manifest = Path.join(stage.directory, "pilot.json")
+
+    if File.exists?(manifest),
+      do: raise("Pilot already ran in this installation; use check-pilot to verify it")
+
+    native = native_check(stage)
+    S.private_json(manifest, %{})
+
+    try do
+      S.run(["node", "scripts/staging-pilot.js", "seed", stage.url, manifest], capture: false)
+    rescue
+      error ->
+        if S.json(manifest) == %{}, do: File.rm!(manifest)
+        reraise error, __STACKTRACE__
+    end
+
+    outage_check(stage)
+    stage = upgrade(stage, stage.settings["SYNIXIR_IMAGE"])
+    check_pilot(stage)
+    restored = verify(stage, backup(stage))
+
+    S.eventually(
+      fn ->
+        targets =
+          compose(stage, [
+            "exec",
+            "-T",
+            "prometheus",
+            "wget",
+            "-qO-",
+            "http://localhost:9090/api/v1/targets"
+          ])
+          |> JSON.decode!()
+          |> get_in(["data", "activeTargets"])
+
+        targets != [] and Enum.all?(targets, &(&1["health"] == "up"))
+      end,
+      "Prometheus did not scrape the private metrics endpoint"
+    )
+
+    report = %{
+      checked_at: S.now(),
+      image: stage.settings["SYNIXIR_IMAGE"],
+      https: true,
+      secure_cookies: true,
+      collaboration_and_roles: true,
+      offline_reconnect: true,
+      same_image_upgrade: true,
+      private_metrics_scraped: true,
+      restored_database: restored,
+      native_libraries: native,
+      database_outage_recovery: true,
+      image_id:
+        S.run([
+          "docker",
+          "image",
+          "inspect",
+          stage.settings["SYNIXIR_IMAGE"],
+          "--format",
+          "{{.Id}}"
+        ])
+        |> String.trim()
+    }
+
+    S.private_json(Path.join(stage.directory, "pilot-result.json"), report)
+    S.print(report)
+  end
+
+  def rehearse(image, port, output) do
+    S.unused!(output)
+    local = Path.join(S.root(), ".local")
+    File.mkdir_p!(local)
+
+    S.temporary(
+      fn temporary ->
+        directory = Path.join(temporary, "stage")
+        initialize(directory, "synixir-pilot-" <> S.token(), port, image)
+        stage = open(directory)
+
+        try do
+          up(stage)
+          pilot(stage)
+          S.private_json(output, S.json(Path.join(directory, "pilot-result.json")))
+        after
+          # Remove only the project and fixture volume created by this invocation.
+          compose(stage, ["down", "--volumes"], capture: false)
+        end
+      end,
+      local
+    )
+  end
+end