diff --git a/DECISIONS/2026-11-09-prod-db-naming.md b/DECISIONS/2026-11-09-prod-db-naming.md
new file mode 100644
index 0000000..30de10b
--- /dev/null
+++ b/DECISIONS/2026-11-09-prod-db-naming.md
@@ -0,0 +1,42 @@
+# Decision: Production database name and owner (#172)
+
+Status:  Decided and applied
+Date:    2026-11-09
+Review:  When the on-prem and Azure targets are unified, or Azure setup (#31, #32, #34) changes the server layout
+
+## Choice
+
+The Azure production database is **`te_prod`, owned by `pgadmin`**. Terraform creates it, and nothing else names it.
+
+- `infra/terraform-prod/main.tf` creates the database `te_prod` on a server whose administrator login is `pgadmin`.
+- The migration job receives them as `PGDATABASE` and `PGUSER`. `infra/entrypoint.sh` writes them into the launcher (`env_prod.sql`) it deploys with, as `db_name` and `db_owner`.
+- `build/deploy_all.sh` reads `db_name` and `db_owner` from the launcher instead of hard-coding `te_mgmt_<env>` and `postgres`. A launcher that sets neither keeps the old names.
+
+The on-prem installer is unchanged: it still deploys `te_mgmt_prod`, owned by `postgres`, from the committed template. So there are two production names, one per target, and each is set in one place.
+
+## Context
+
+#172: Terraform created `te_prod` owned by `pgadmin`; the deployer created `te_mgmt_prod` owned by `postgres`; the entrypoint expected an `env_prod.sql` that nothing in the image produced. A fresh image failed at the first step (`Environment file not found`), and had it got past that, it would have created a second database on Azure, where no `postgres` role exists.
+
+## Reasoning
+
+- Azure Database for PostgreSQL has no `postgres` superuser. The deployer's `OWNER = postgres` cannot work there; Terraform's `pgadmin` is the only owner available.
+- The launcher is already the file that names the database. Reading the name from it makes the deployer agree with whatever the launcher says, rather than with a second copy of the name.
+- The launcher holds a credential, so it must not be in the image (#174). Generating it at run time from the container's own settings removes the missing file without baking anything in. The application password is read by psql from `APP_DB_PASSWORD` (`\getenv`), so it never reaches a file.
+- Renaming prod to `te_prod` everywhere was considered and rejected for now. `te_mgmt_prod` is used by the offline installer (`scripts/offline/install.sh`, #60 has a 15 Nov deadline), the prod environment tests, the evals runner, CI and several docs. The change would touch about 15 files and a procedure that is meant to ship.
+
+## Also changed, because the first deployment needs it
+
+- Terraform allow-lists `dblink`, `pg_trgm` and `uuid-ossp` (`azure.extensions`). Azure refuses `CREATE EXTENSION` for anything not on that list, and the schema creates all three, so the first deployment would otherwise have failed partway through.
+- Terraform creates the application role's password (`pg-app-password` in Key Vault) and passes it to the job as `APP_DB_PASSWORD`. The runner identity already reads secrets from that vault.
+- `build/te_core_schema.sql` no longer prints the application password. A `SELECT set_config(..., :'app_password', ...)` echoed it into the deployment output, which a container job sends to its logs. It now uses `\gset`, which keeps the row out of the output.
+
+## Trade-offs accepted
+
+- Two production names exist until the targets are unified.
+- `te_prod` is also the schema name inside the database, so the Azure database and schema share a name (`te_prod.te_prod`). It is harmless, and matches what Terraform already created.
+- Verified without a Docker daemon: the entrypoint and deployer were run directly against a fresh PostgreSQL 16 whose only role is `pgadmin`. An image build and a run on Azure itself have not been done.
+
+## Escape hatch
+
+Mount a concrete `env_prod.sql` over the generated one; the entrypoint leaves an existing launcher alone.
diff --git a/DECISIONS/2026-11-09-tenant-te-migrations.md b/DECISIONS/2026-11-09-tenant-te-migrations.md
new file mode 100644
index 0000000..bb22148
--- /dev/null
+++ b/DECISIONS/2026-11-09-tenant-te-migrations.md
@@ -0,0 +1,38 @@
+# Decision: Versioned migrations for tenant T&E tables (#184)
+
+Status:  Decided and applied
+Date:    2026-11-09
+Review:  When there are live tenants, or when a second T&E schema change is made (does the two-copy cost hold up?)
+
+## Choice
+
+The twelve T&E tables in each tenant schema get their own Alembic chain, `alembic_te/`, with its version table `alembic_version_te` inside the tenant schema. `migrate_tenant` clones a new tenant from the template and stamps it, adopts a tenant from before versioning at a baseline (`te0001`), and upgrades every tenant to head. A parity test proves a migrated tenant equals the template.
+
+Chosen by the owner on 9 Nov from three options:
+
+- **Alembic chain + parity test** (this)
+- Apply `te_core_schema.sql` to tenants (one source of truth, but it is a psql script that creates databases and roles, and the API would need psql at run time)
+- Reconcile from the template (smallest, but cannot backfill `program_id` or replace foreign keys safely)
+
+## Context
+
+`provision_te_tables` cloned only missing tables, so a template change never reached a tenant that already had them. #177 (composite foreign keys, `program_id`) made that concrete: every tenant created before it kept the old shape and the old protection against cross-program rows.
+
+## Reasoning
+
+- #177 is a data-aware change (backfill, a check that can fail on existing rows, replacing constraints). Only a migration can do that safely and say what it found.
+- A tenant needs to know which version it is at. A version table per tenant schema matches how the uploads chain already works.
+- The template is still what new tenants are cloned from, so provisioning stays cheap; the chain only has to carry *changes*. The baseline adopts what exists rather than re-creating the tables, so there is no second copy of the original DDL.
+
+## Trade-offs accepted
+
+- **Each T&E change is written twice**: in `te_core_schema.sql` (both copies) and as a migration. The parity test fails on any difference, which is the guard; there is no generator. See `docs/TENANT_SCHEMA_MIGRATIONS.md`.
+- A failing tenant stops API start (as the uploads chain does for duplicate registry rows). One tenant's bad data takes the API down until it is fixed. The upgrade rolls back, so nothing is half-applied, and the error names the tenant and the rows. A per-tenant "needs attention" state was not built.
+- The "old shape" test fixture comes from the chain's own `downgrade`, not an independent dump of the pre-#177 schema. The test asserts what that shape is, so the downgrade is not taken on trust.
+- Tenants cloned by the old provisioning keep default index names on indexes the migrations don't touch (listed in the doc). Only names.
+
+## Also changed, found by the parity test
+
+- `CREATE TABLE ... LIKE ... INCLUDING ALL` names copied indexes and unique constraints by PostgreSQL's defaults, not the template's, so every tenant already differed from its template in index names, and `te0002`'s `uq_*` constraints would have been added a second time beside the default-named ones. Provisioning now renames the copies to the template's names, and `te0002` adopts an equivalent unique constraint instead of adding another.
+- The runtime role cannot rewrite `alembic_version_te` (added to `APPEND_ONLY`, like `alembic_version`).
+- The template schema carries `COMMENT ON SCHEMA ... 'te_schema_version=te0002'`; a test fails if a migration is added and the stamp is left behind.
diff --git a/alembic/versions/0009_type_provenance.py b/alembic/versions/0009_type_provenance.py
new file mode 100644
index 0000000..75a573e
--- /dev/null
+++ b/alembic/versions/0009_type_provenance.py
@@ -0,0 +1,41 @@
+"""Record whether a dynamic table's column types were confirmed by a person (#195).
+
+Decided 4 Nov: a dynamic-mode table is created from types a person confirmed,
+and the registry says which. ``types_provenance`` is ``user_confirmed`` or
+``inferred_unconfirmed``. T&E rows keep NULL: their columns are the fixed
+schema's, so there are no inferred types to confirm.
+
+Dynamic rows registered before this revision are marked ``inferred_unconfirmed``:
+nothing recorded whether anyone confirmed their types, and the flag must not
+claim a confirmation that can't be shown. Runs per tenant.
+
+Revision ID: 0009
+Revises: 0008
+Create Date: 2026-11-09
+"""
+
+from alembic import op
+
+from api.config import settings
+
+revision = "0009"
+down_revision = "0008"
+branch_labels = None
+depends_on = None
+
+
+def _schema() -> str:
+    return op.get_bind().dialect.identifier_preparer.quote(settings.UPLOADS_SCHEMA)
+
+
+def upgrade() -> None:
+    s = _schema()
+    op.execute(f"ALTER TABLE {s}.csv_files ADD COLUMN IF NOT EXISTS types_provenance TEXT "
+               "CHECK (types_provenance IN ('inferred_unconfirmed', 'user_confirmed'))")
+    # s is the dialect-quoted schema identifier from settings, never request input.
+    op.execute(f"UPDATE {s}.csv_files SET types_provenance = 'inferred_unconfirmed' "  # nosec B608
+               "WHERE mode = 'dynamic' AND types_provenance IS NULL")
+
+
+def downgrade() -> None:
+    op.execute(f"ALTER TABLE {_schema()}.csv_files DROP COLUMN IF EXISTS types_provenance")
diff --git a/alembic_te/env.py b/alembic_te/env.py
new file mode 100644
index 0000000..5b4e060
--- /dev/null
+++ b/alembic_te/env.py
@@ -0,0 +1,46 @@
+"""Alembic environment for the T&E tables inside ONE tenant schema (#184).
+
+A second chain, separate from ``alembic/``: that one versions the uploads objects
+(``csv_files``, ``audit_log``, ...); this one versions the twelve fixed T&E tables
+that api/tenancy.py clones into each tenant from the deployment's template schema.
+Its version table is ``alembic_version_te``, inside the tenant schema, so each tenant
+carries its own T&E schema version.
+
+It never runs from a command line: the target schema comes from the caller
+(``cfg.attributes["schema"]``), and api/tenancy.py serialises callers per tenant.
+"""
+
+import sys
+from pathlib import Path
+
+from alembic import context
+from sqlalchemy import create_engine
+from sqlalchemy.engine import URL
+from sqlalchemy.pool import NullPool
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from api.config import settings  # noqa: E402
+
+VERSION_TABLE = "alembic_version_te"
+
+if context.is_offline_mode():
+    raise RuntimeError("Offline (--sql) migrations are not supported; run against a database.")
+
+schema = context.config.attributes.get("schema")
+if not schema:
+    raise RuntimeError("T&E migrations run against one tenant schema. Use api.tenancy.migrate_tenant(slug).")
+
+url = URL.create(
+    "postgresql+psycopg2",
+    username=settings.PG_USER,
+    password=settings.PG_PASSWORD or None,
+    host=settings.PG_HOST,
+    port=settings.PG_PORT,
+    database=settings.PG_DATABASE,
+)
+engine = create_engine(url, poolclass=NullPool)
+with engine.connect() as connection:
+    context.configure(connection=connection, version_table=VERSION_TABLE, version_table_schema=schema)
+    with context.begin_transaction():
+        context.run_migrations()
diff --git a/alembic_te/script.py.mako b/alembic_te/script.py.mako
new file mode 100644
index 0000000..417d68f
--- /dev/null
+++ b/alembic_te/script.py.mako
@@ -0,0 +1,21 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+"""
+
+from alembic import op
+
+revision = ${repr(up_revision)}
+down_revision = ${repr(down_revision)}
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    ${downgrades if downgrades else "pass"}
diff --git a/alembic_te/versions/te0001_baseline.py b/alembic_te/versions/te0001_baseline.py
new file mode 100644
index 0000000..72b3cd5
--- /dev/null
+++ b/alembic_te/versions/te0001_baseline.py
@@ -0,0 +1,24 @@
+"""Baseline: the T&E tables as they were before versioned tenant migrations (#184).
+
+Nothing to run. This revision names the shape every tenant had until now: the
+twelve tables cloned from the template by api/tenancy.py's ``provision_te_tables``,
+before #177 added ``program_id`` and the composite foreign keys. A tenant that has
+T&E tables but no version table is stamped here, then upgraded to head.
+
+Revision ID: te0001
+Revises:
+Create Date: 2026-11-09
+"""
+
+revision = "te0001"
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass
diff --git a/alembic_te/versions/te0002_program_consistency.py b/alembic_te/versions/te0002_program_consistency.py
new file mode 100644
index 0000000..774352b
--- /dev/null
+++ b/alembic_te/versions/te0002_program_consistency.py
@@ -0,0 +1,237 @@
+"""Program consistency for the T&E tables (#177), applied to existing tenants (#184).
+
+Ports phase 4B of build/te_core_schema.sql, which does the same for a template
+schema. A tenant's tables are cloned from the template, so a tenant created before
+#177 kept the old shape for good; this brings it across:
+
+- ``program_id`` on test_cases, test_events, vcrm_entries and test_results, backfilled
+  from the parent row each already depends on (``updated_at`` triggers paused, so the
+  backfill does not look like an edit), then NOT NULL;
+- every cross-table foreign key becomes composite ``(<id>, program_id)``, backed by
+  UNIQUE (<id>, program_id), so PostgreSQL rejects a row that joins two programs;
+- ``te_fill_program_id`` and its BEFORE INSERT triggers fill ``program_id`` from the
+  parent when a load omits it.
+
+Idempotent: a tenant cloned after #177 already has all of it, and re-running changes
+nothing. The whole upgrade is one transaction. If cross-program rows already exist it
+ABORTS, leaving the tenant exactly as it was, and lists them so they can be triaged.
+
+The names below must match phase 4B; tests/system/test_tenant_te_migrations.py compares a
+migrated tenant with the template and fails if they drift.
+
+Revision ID: te0002
+Revises: te0001
+Create Date: 2026-11-09
+"""
+
+from alembic import op
+from psycopg2 import sql
+
+revision = "te0002"
+down_revision = "te0001"
+branch_labels = None
+depends_on = None
+
+# Body copied byte for byte from phase 4B: pg_get_functiondef must match the template's.
+_FILL_FUNCTION_BODY = """$$
+-- args: parent table, parent key column, this row's FK column. The parent lives
+-- in the row's own schema (TG_TABLE_SCHEMA), never a name baked in at deploy
+-- time: api/tenancy.py clones these triggers into per-tenant schemas.
+DECLARE
+   v_prog UUID;
+BEGIN
+   IF NEW.program_id IS NULL THEN
+      EXECUTE format('SELECT program_id FROM %I.%I WHERE %I = $1',
+                     TG_TABLE_SCHEMA, TG_ARGV[0], TG_ARGV[1])
+         INTO v_prog USING (to_jsonb(NEW) ->> TG_ARGV[2])::UUID;
+      IF v_prog IS NULL THEN
+         RAISE EXCEPTION '%.% % not found', TG_ARGV[0], TG_ARGV[1], to_jsonb(NEW) ->> TG_ARGV[2]
+            USING ERRCODE = 'foreign_key_violation';
+      END IF;
+      NEW.program_id := v_prog;
+   END IF;
+   RETURN NEW;
+END;
+$$"""
+
+_PROGRAM_TABLES = ("test_cases", "test_events", "vcrm_entries", "test_results")
+
+# (table, constraint, definition) — UNIQUE targets first, then the composite FKs that need them.
+_UNIQUES = (
+    ("test_phases", "uq_phases_id_program", "UNIQUE (phase_id, program_id)"),
+    ("requirements", "uq_requirements_id_program", "UNIQUE (req_id, program_id)"),
+    ("test_cases", "uq_testcases_id_program", "UNIQUE (tc_id, program_id)"),
+    ("test_events", "uq_events_id_program", "UNIQUE (event_id, program_id)"),
+    ("test_results", "uq_results_id_program", "UNIQUE (result_id, program_id)"),
+)
+# (table, constraint, columns, parent, ON DELETE clause)
+_COMPOSITE_FKS = (
+    ("test_cases", "fk_testcases_phase_program", "phase_id", "test_phases", " ON DELETE CASCADE"),
+    ("test_events", "fk_events_phase_program", "phase_id", "test_phases", " ON DELETE CASCADE"),
+    ("vcrm_entries", "fk_vcrm_req_program", "req_id", "requirements", " ON DELETE CASCADE"),
+    ("vcrm_entries", "fk_vcrm_tc_program", "tc_id", "test_cases", " ON DELETE CASCADE"),
+    ("test_results", "fk_results_event_program", "event_id", "test_events", " ON DELETE CASCADE"),
+    ("test_results", "fk_results_tc_program", "tc_id", "test_cases", ""),
+    # result_id is nullable; MATCH SIMPLE skips the check when it is NULL
+    ("defect_reports", "fk_defects_result_program", "result_id", "test_results", ""),
+)
+# (table, parent, parent key, this row's FK column) for the fill triggers
+_FILL_TRIGGERS = (
+    ("test_cases", "test_phases", "phase_id", "phase_id"),
+    ("test_events", "test_phases", "phase_id", "phase_id"),
+    ("vcrm_entries", "requirements", "req_id", "req_id"),
+    ("test_results", "test_events", "event_id", "event_id"),
+)
+# The single-column FKs the composites replace, and what they were before (for downgrade).
+_SINGLE_FKS = (
+    ("test_cases", "phase_id", "test_phases", " ON DELETE CASCADE"),
+    ("test_events", "phase_id", "test_phases", " ON DELETE CASCADE"),
+    ("vcrm_entries", "req_id", "requirements", " ON DELETE CASCADE"),
+    ("vcrm_entries", "tc_id", "test_cases", " ON DELETE CASCADE"),
+    ("test_results", "event_id", "test_events", " ON DELETE CASCADE"),
+    ("test_results", "tc_id", "test_cases", ""),
+    ("defect_reports", "result_id", "test_results", ""),
+)
+
+
+def _schema() -> str:
+    return op.get_context().opts["version_table_schema"]
+
+
+def _cursor():
+    return op.get_bind().connection.cursor()
+
+
+def _tbl(schema: str, table: str) -> sql.Composable:
+    return sql.SQL("{}.{}").format(sql.Identifier(schema), sql.Identifier(table))
+
+
+def _has_constraint(cur, schema: str, table: str, name: str) -> bool:
+    cur.execute(sql.SQL("SELECT 1 FROM pg_constraint WHERE conrelid = {}::regclass AND conname = %s").format(
+        sql.Literal(f'"{schema}"."{table}"')), (name,))
+    return cur.fetchone() is not None
+
+
+def _add_constraint(cur, schema: str, table: str, name: str, definition: sql.Composable | str) -> None:
+    if _has_constraint(cur, schema, table, name):
+        return
+    if isinstance(definition, str) and definition.startswith("UNIQUE"):
+        # A tenant cloned after #177 but before provisioning kept the template's names has this
+        # constraint under a default name. Adopt it; a second copy would be a redundant index.
+        cur.execute("SELECT conname FROM pg_constraint WHERE conrelid = %s::regclass AND contype = 'u' "
+                    "AND pg_get_constraintdef(oid) = %s", (f'"{schema}"."{table}"', definition))
+        row = cur.fetchone()
+        if row:
+            cur.execute(sql.SQL("ALTER TABLE {} RENAME CONSTRAINT {} TO {}").format(
+                _tbl(schema, table), sql.Identifier(row[0]), sql.Identifier(name)))
+            return
+    cur.execute(sql.SQL("ALTER TABLE {} ADD CONSTRAINT {} {}").format(
+        _tbl(schema, table), sql.Identifier(name),
+        definition if isinstance(definition, sql.Composable) else sql.SQL(definition)))
+
+
+def _drop_constraint(cur, schema: str, table: str, name: str, if_exists: bool = False) -> None:
+    cur.execute(sql.SQL("ALTER TABLE {} DROP CONSTRAINT {}{}").format(
+        _tbl(schema, table), sql.SQL("IF EXISTS " if if_exists else ""), sql.Identifier(name)))
+
+
+def _toggle_updated_at(cur, schema: str, tables: tuple[str, ...], action: str) -> None:
+    for table in tables:
+        cur.execute(sql.SQL("SELECT 1 FROM pg_trigger WHERE tgrelid = {}::regclass AND tgname = 'trg_updated_at'")
+                    .format(sql.Literal(f'"{schema}"."{table}"')))
+        if cur.fetchone():
+            statement = sql.SQL("ALTER TABLE {} {} TRIGGER trg_updated_at")
+            cur.execute(statement.format(_tbl(schema, table), sql.SQL(action)))
+
+
+def upgrade() -> None:
+    schema, cur = _schema(), _cursor()
+    t = {name: _tbl(schema, name) for name in
+         ("test_phases", "requirements", "test_cases", "vcrm_entries", "test_events", "test_results",
+          "defect_reports")}
+
+    cur.execute(sql.SQL("CREATE OR REPLACE FUNCTION {}.te_fill_program_id() RETURNS TRIGGER LANGUAGE plpgsql AS ")
+                .format(sql.Identifier(schema)) + sql.SQL(_FILL_FUNCTION_BODY))
+
+    # 1. program_id columns, nullable until backfilled
+    for table in _PROGRAM_TABLES:
+        cur.execute(sql.SQL("ALTER TABLE {} ADD COLUMN IF NOT EXISTS program_id UUID").format(t[table]))
+
+    # 2. backfill from the parent each row already depends on
+    _toggle_updated_at(cur, schema, ("test_cases", "test_events"), "DISABLE")
+    for child, parent, key in (("test_cases", "test_phases", "phase_id"),
+                               ("test_events", "test_phases", "phase_id"),
+                               ("vcrm_entries", "requirements", "req_id"),
+                               ("test_results", "test_events", "event_id")):
+        cur.execute(sql.SQL("UPDATE {c} c SET program_id = p.program_id FROM {p} p "
+                            "WHERE c.{k} = p.{k} AND c.program_id IS NULL").format(
+                                c=t[child], p=t[parent], k=sql.Identifier(key)))
+    _toggle_updated_at(cur, schema, ("test_cases", "test_events"), "ENABLE")
+
+    # 3. refuse to constrain data that already violates BR-07, and say which rows to triage
+    cur.execute(sql.SQL("""
+        SELECT string_agg(msg, E'\\n') FROM (
+           SELECT 'vcrm_entries ' || v.vcrm_id || ': requirement in program ' || v.program_id
+                  || ', test case in program ' || c.program_id AS msg
+           FROM {vc} v JOIN {tc} c ON c.tc_id = v.tc_id WHERE c.program_id <> v.program_id
+           UNION ALL
+           SELECT 'test_results ' || x.result_id || ': event in program ' || x.program_id
+                  || ', test case in program ' || c.program_id
+           FROM {rs} x JOIN {tc} c ON c.tc_id = x.tc_id WHERE c.program_id <> x.program_id
+           UNION ALL
+           SELECT 'defect_reports ' || d.defect_id || ': defect in program ' || d.program_id
+                  || ', result in program ' || x.program_id
+           FROM {df} d JOIN {rs} x ON x.result_id = d.result_id WHERE x.program_id <> d.program_id
+           LIMIT 50
+        ) m""").format(vc=t["vcrm_entries"], tc=t["test_cases"], rs=t["test_results"], df=t["defect_reports"]))
+    (bad,) = cur.fetchone()
+    if bad:
+        raise RuntimeError(f"Tenant schema {schema!r}: cross-program references found (BR-07). "
+                           f"Nothing was changed. Fix or remove these rows, then re-run:\n{bad}")
+
+    # 4. NOT NULL, composite-key targets, composite FKs
+    for table in _PROGRAM_TABLES:
+        cur.execute(sql.SQL("ALTER TABLE {} ALTER COLUMN program_id SET NOT NULL").format(t[table]))
+
+    # drop the single-column FKs the composites replace, whatever they are called
+    for table, column, _parent, _action in _SINGLE_FKS:
+        cur.execute(
+            "SELECT c.conname FROM pg_constraint c JOIN pg_attribute a "
+            "ON a.attrelid = c.conrelid AND a.attnum = c.conkey[1] "
+            "WHERE c.contype = 'f' AND cardinality(c.conkey) = 1 AND c.conrelid = %s::regclass "
+            "AND a.attname = %s", (f'"{schema}"."{table}"', column))
+        for (conname,) in cur.fetchall():
+            _drop_constraint(cur, schema, table, conname)
+
+    for table, name, definition in _UNIQUES:
+        _add_constraint(cur, schema, table, name, definition)
+    for table, name, column, parent, on_delete in _COMPOSITE_FKS:
+        _add_constraint(cur, schema, table, name, sql.SQL(
+            "FOREIGN KEY ({col}, program_id) REFERENCES {parent}({col}, program_id){action}").format(
+                col=sql.Identifier(column), parent=_tbl(schema, parent), action=sql.SQL(on_delete)))
+
+    # 5. fill program_id from the parent on INSERT when the caller omitted it
+    for table, parent, key, fk in _FILL_TRIGGERS:
+        cur.execute(sql.SQL("DROP TRIGGER IF EXISTS trg_fill_program_id ON {}").format(t[table]))
+        cur.execute(sql.SQL("CREATE TRIGGER trg_fill_program_id BEFORE INSERT ON {} FOR EACH ROW "
+                            "EXECUTE FUNCTION {}.te_fill_program_id({}, {}, {})").format(
+                                t[table], sql.Identifier(schema), sql.Literal(parent), sql.Literal(key),
+                                sql.Literal(fk)))
+
+
+def downgrade() -> None:
+    """Back to the shape before #177. Used by tests, and to roll back a bad release."""
+    schema, cur = _schema(), _cursor()
+    for table, *_ in _FILL_TRIGGERS:
+        cur.execute(sql.SQL("DROP TRIGGER IF EXISTS trg_fill_program_id ON {}").format(_tbl(schema, table)))
+    for table, name, *_ in _COMPOSITE_FKS:
+        _drop_constraint(cur, schema, table, name, if_exists=True)
+    for table, column, parent, on_delete in _SINGLE_FKS:
+        cur.execute(sql.SQL("ALTER TABLE {} ADD FOREIGN KEY ({}) REFERENCES {}({}){}").format(
+            _tbl(schema, table), sql.Identifier(column), _tbl(schema, parent), sql.Identifier(column),
+            sql.SQL(on_delete)))
+    for table, name, _definition in _UNIQUES:
+        _drop_constraint(cur, schema, table, name, if_exists=True)
+    for table in _PROGRAM_TABLES:
+        cur.execute(sql.SQL("ALTER TABLE {} DROP COLUMN IF EXISTS program_id").format(_tbl(schema, table)))
+    cur.execute(sql.SQL("DROP FUNCTION IF EXISTS {}.te_fill_program_id()").format(sql.Identifier(schema)))
diff --git a/api/db.py b/api/db.py
index 4f36d96..8ffe71d 100644
--- a/api/db.py
+++ b/api/db.py
@@ -203,7 +203,7 @@ def owner_connect():
 
 # Tables the runtime role may only read and insert (M3, #107). Triggers also stop the owner.
 APPEND_ONLY = ("audit_log", "uploads", "upload_blobs", "source_rows", "row_decisions", "cell_edits",
-               "alembic_version")
+               "alembic_version", "alembic_version_te")
 
 
 def grant_runtime(cur, schema: str, create: bool) -> None:
diff --git a/api/routers/csv_routes.py b/api/routers/csv_routes.py
index f4d5192..2c9b16e 100644
--- a/api/routers/csv_routes.py
+++ b/api/routers/csv_routes.py
@@ -53,6 +53,8 @@ class UploadRequest(BaseModel):
     # The file's bytes as the browser read them, base64. Kept as the preserved
     # original so its hash matches the uploader's copy; `content` is still what loads (#185).
     contentBase64: str | None = None
+    # True when a person reviewed `types` before this call (dynamic mode only, #195).
+    typesConfirmed: bool = False
 
 
 @router.post("/preview", dependencies=[Depends(allow("preview"))])
@@ -104,11 +106,15 @@ def upload(req: UploadRequest, session: dict | None = Depends(tenant_scope)) ->
             # T&E columns must match the fixed schema exactly; a rename could
             # only be ignored, which is the silent failure #186 describes.
             raise HTTPException(422, "headerMapping is only supported in 'dynamic' mode")
+        if req.typesConfirmed:
+            raise HTTPException(422, "typesConfirmed is only supported in 'dynamic' mode")
         result = upload_te(req.fileName, req.content, req.targetTable, session=session)
     elif req.mode == "dynamic":
+        if req.typesConfirmed and not req.types:
+            raise HTTPException(422, "typesConfirmed requires the confirmed types to be sent in 'types'")
         mapping = [m.model_dump() for m in req.headerMapping] if req.headerMapping else None
         result = upload_dynamic(req.fileName, req.content, req.types, req.overwrite,
-                                session=session, header_mapping=mapping)
+                                session=session, header_mapping=mapping, types_confirmed=req.typesConfirmed)
     else:
         raise HTTPException(422, f"Unknown mode '{req.mode}' — use 'dynamic' or 'te'")
 
@@ -138,7 +144,7 @@ def list_files() -> list[dict]:
             cur.execute(
                 sql.SQL(
                     "SELECT f.id, f.file_name, f.table_name, f.mode, f.row_count, f.column_names, f.created_at, "
-                    "f.total_rows, f.duplicate_rows_skipped, f.failed_rows, u.uploader, u.id "
+                    "f.total_rows, f.duplicate_rows_skipped, f.failed_rows, u.uploader, u.id, f.types_provenance "
                     "FROM {s}.csv_files f LEFT JOIN LATERAL (SELECT uploader, id FROM {s}.uploads "
                     "  WHERE csv_file_id = f.id ORDER BY id DESC LIMIT 1) u ON true "
                     "ORDER BY f.created_at DESC, f.id DESC"
@@ -158,6 +164,7 @@ def list_files() -> list[dict]:
                     "failed_rows": r[9],
                     "uploader": r[10],
                     "upload_id": None if r[11] is None else str(r[11]),
+                    "types_provenance": r[12],
                 }
                 for r in cur.fetchall()
             ]
diff --git a/api/services/dynamic_loader.py b/api/services/dynamic_loader.py
index b8464d2..8e93982 100644
--- a/api/services/dynamic_loader.py
+++ b/api/services/dynamic_loader.py
@@ -60,6 +60,7 @@ def upload_dynamic(
     *,
     session: dict | None = None,
     header_mapping: list[dict] | None = None,
+    types_confirmed: bool = False,
 ) -> dict:
     logs: list[dict] = []
     _log(logs, "receive", f'Received "{file_name}" ({len(content)} chars)')
@@ -122,9 +123,13 @@ def upload_dynamic(
         }
 
     schema = uploads_schema()
+    # #195: only types a person explicitly confirmed count as confirmed; omitted types
+    # fall back to text and a bare `types` list is not proof anyone looked at it.
+    provenance = "user_confirmed" if (types and types_confirmed) else "inferred_unconfirmed"
 
     try:
-        return _do_upload(file_name, types, overwrite, logs, schema, rows, file_hash, session=session)
+        return _do_upload(file_name, types, overwrite, logs, schema, rows, file_hash,
+                          session=session, types_provenance=provenance)
     except psycopg2.errors.UniqueViolation:
         # #81: an identical concurrent upload won a race that _do_upload could
         # not settle itself (e.g. both creating csv_<hash> at once). Nothing
@@ -161,6 +166,7 @@ def _do_upload(
     file_hash: str,
     *,
     session: dict | None = None,
+    types_provenance: str = "inferred_unconfirmed",
 ) -> dict:
     replaced_file_name: str | None = None
     with Conn() as conn:
@@ -369,11 +375,11 @@ def _do_upload(
                 cur.execute(
                     sql.SQL(
                         "INSERT INTO {}.csv_files (file_name, file_hash, table_name, mode, row_count, column_names, "
-                        "total_rows, duplicate_rows_skipped, failed_rows) "
-                        "VALUES (%s, %s, %s, 'dynamic', %s, %s, %s, %s, %s) RETURNING id"
+                        "total_rows, duplicate_rows_skipped, failed_rows, types_provenance) "
+                        "VALUES (%s, %s, %s, 'dynamic', %s, %s, %s, %s, %s, %s) RETURNING id"
                     ).format(sql.Identifier(schema)),
                     (file_name, file_hash, table_name, inserted, columns,
-                     len(data_rows), duplicates, failed_row_count),
+                     len(data_rows), duplicates, failed_row_count, types_provenance),
                 )
                 file_id = cur.fetchone()[0]
             except psycopg2.errors.UniqueViolation:
@@ -409,6 +415,7 @@ def _do_upload(
         "failedRows": failed_row_count,
         "columns": columns,
         "types": col_types,
+        "typesProvenance": types_provenance,
         "rowErrors": row_errors,
         "logs": logs,
         "overwritten": replaced_file_name is not None,
diff --git a/api/tenancy.py b/api/tenancy.py
index bf73824..2373083 100644
--- a/api/tenancy.py
+++ b/api/tenancy.py
@@ -8,9 +8,11 @@
 
 from __future__ import annotations
 
+import re
 import threading
 from contextlib import contextmanager
 from contextvars import ContextVar
+from pathlib import Path
 
 from psycopg2 import sql
 
@@ -70,11 +72,132 @@ def migrate_tenant(slug: str) -> None:
 
     with owner_connect() as conn:
         with conn.cursor() as cur:
-            provision_te_tables(cur, schema_for(slug), settings.TE_SCHEMA)
-            conn.commit()
+            migrate_te_schema(conn, cur, schema_for(slug))
             grant_runtime(cur, schema_for(slug), create=True)
 
 
+TE_VERSION_TABLE = "alembic_version_te"
+TE_BASELINE = "te0001"
+_TE_VERSION_COMMENT = re.compile(r"te_schema_version=(\w+)")
+
+
+def _te_config(schema: str):
+    from alembic.config import Config
+
+    cfg = Config()
+    cfg.set_main_option("script_location", str(Path(__file__).resolve().parent.parent / "alembic_te"))
+    cfg.attributes["schema"] = schema
+    return cfg
+
+
+def te_head() -> str:
+    """The newest T&E migration this release knows (build/te_core_schema.sql must stamp the same)."""
+    from alembic.script import ScriptDirectory
+
+    return ScriptDirectory.from_config(_te_config("unused")).get_current_head()
+
+
+def template_te_version(cur, template: str) -> str | None:
+    """The T&E migration the template schema is at, as build/te_core_schema.sql stamped it."""
+    cur.execute("SELECT obj_description(oid, 'pg_namespace') FROM pg_namespace WHERE nspname = %s", (template,))
+    row = cur.fetchone()
+    m = _TE_VERSION_COMMENT.search(row[0] or "") if row else None
+    return m.group(1) if m else None
+
+
+def _te_tables_present(cur, schema: str) -> set[str]:
+    cur.execute("SELECT c.relname FROM pg_class c WHERE c.relnamespace = to_regnamespace(%s) "
+                "AND c.relkind = 'r' AND c.relname = ANY(%s)", (schema, TE_TABLES))
+    return {r[0] for r in cur.fetchall()}
+
+
+def migrate_te_schema(conn, cur, schema: str) -> str:
+    """Bring one tenant's T&E tables to the newest migration (#184). Returns what was done.
+
+    - no T&E tables: clone them from the template (#99), stamp the tenant at the template's
+      version, and upgrade from there;
+    - tables but no version table (a tenant from before versioned migrations): stamp the
+      baseline, then upgrade;
+    - otherwise: upgrade.
+
+    Each tenant's migrations are one transaction; one that fails (for instance on
+    cross-program rows, which te0002 lists) leaves that tenant as it was. Serialised per
+    tenant across processes by an advisory lock, so two API workers starting together
+    cannot both provision or upgrade the same schema. Commits.
+    """
+    from alembic import command
+    from alembic.util import CommandError
+
+    cfg, template = _te_config(schema), settings.TE_SCHEMA
+    cur.execute("SELECT pg_advisory_lock(hashtextextended(%s, 0))", (f"te_schema:{schema}",))
+    try:
+        present = _te_tables_present(cur, schema)
+        cur.execute("SELECT to_regclass(%s)", (f'"{schema}"."{TE_VERSION_TABLE}"',))
+        versioned = cur.fetchone()[0] is not None
+        conn.commit()
+        if versioned and len(present) == len(TE_TABLES):
+            action, stamp = "upgraded", None
+        elif not present:
+            template_version = template_te_version(cur, template)
+            if template_version is None:
+                raise RuntimeError(
+                    f"T&E template schema {template!r} has no te_schema_version stamp; it was deployed by an "
+                    "older build/te_core_schema.sql. Redeploy it with build/deploy_all.sh first.")
+            provision_te_tables(cur, schema, template)
+            conn.commit()
+            action, stamp = "provisioned", template_version
+        else:
+            if len(present) < len(TE_TABLES):
+                provision_te_tables(cur, schema, template)
+                conn.commit()
+            action, stamp = "adopted", None if versioned else TE_BASELINE
+        try:
+            with _migrate_lock:  # Alembic's op/context are process-global; the advisory lock covers other processes
+                if stamp:
+                    command.stamp(cfg, stamp)
+                command.upgrade(cfg, "head")
+        except CommandError as exc:
+            raise RuntimeError(f"Tenant schema {schema!r}: {exc}. If the template is newer than this release, "
+                               "deploy the matching release; if older, redeploy it with "
+                               "build/deploy_all.sh.") from exc
+        return action
+    finally:
+        conn.rollback()
+        cur.execute("SELECT pg_advisory_unlock(hashtextextended(%s, 0))", (f"te_schema:{schema}",))
+        conn.commit()
+
+
+def _use_template_names(cur, schema: str, template: str, tables: list[str]) -> None:
+    """Give the cloned indexes and unique constraints the template's names (#184).
+
+    ``CREATE TABLE ... LIKE ... INCLUDING ALL`` names them by PostgreSQL's default rules
+    (``test_cases_phase_id_idx``), not the template's (``idx_testcases_phase``). A migration
+    that drops or renames an index by name would then work on the template and fail on every
+    tenant. Matched on what each index is (table, columns, method), since the names differ.
+    """
+    def listing(sch: str) -> list[tuple]:
+        cur.execute(
+            "SELECT t.relname, i.relname, pg_get_indexdef(i.oid), c.conname FROM pg_index x "
+            "JOIN pg_class i ON i.oid = x.indexrelid JOIN pg_class t ON t.oid = x.indrelid "
+            "LEFT JOIN pg_constraint c ON c.conindid = x.indexrelid AND c.contype = 'u' "
+            "WHERE t.relnamespace = %s::regnamespace AND t.relname = ANY(%s) AND NOT x.indisprimary",
+            (sch, tables))
+        return [(t, i, re.sub(r"^CREATE (UNIQUE )?INDEX \S+ ON \S+ ", r"\1", d), con)
+                for t, i, d, con in cur.fetchall()]
+
+    mine = {(t, tail): (i, con) for t, i, tail, con in listing(schema)}
+    for table, name, tail, constraint in listing(template):
+        have = mine.get((table, tail))
+        if have is None or have[0] == name:
+            continue
+        if constraint and have[1]:  # renaming a unique constraint renames its index too
+            cur.execute(sql.SQL("ALTER TABLE {}.{} RENAME CONSTRAINT {} TO {}").format(
+                sql.Identifier(schema), sql.Identifier(table), sql.Identifier(have[1]), sql.Identifier(constraint)))
+        elif not constraint and not have[1]:
+            cur.execute(sql.SQL("ALTER INDEX {}.{} RENAME TO {}").format(
+                sql.Identifier(schema), sql.Identifier(have[0]), sql.Identifier(name)))
+
+
 def provision_te_tables(cur, schema: str, template: str) -> int:
     """Give a tenant its own T&E tables, cloned from the deployment's T&E schema (#99).
 
@@ -101,6 +224,7 @@ def provision_te_tables(cur, schema: str, template: str) -> int:
     for table in missing:
         cur.execute(sql.SQL("CREATE TABLE {}.{} (LIKE {}.{} INCLUDING ALL)").format(
             sql.Identifier(schema), sql.Identifier(table), sql.Identifier(template), sql.Identifier(table)))
+    _use_template_names(cur, schema, template, missing)
 
     def defs(query: str) -> list[tuple]:
         cur.execute(query, (template, missing))
diff --git a/build/deploy_all.sh b/build/deploy_all.sh
index be338e9..df6b173 100644
--- a/build/deploy_all.sh
+++ b/build/deploy_all.sh
@@ -5,6 +5,8 @@
 #   ./deploy_all.sh                        # deploy all 4 environments
 #   ./deploy_all.sh dev test               # deploy dev and test only
 #   PGHOST=myserver ./deploy_all.sh prod   # target a specific host
+#   ./deploy_all.sh --dry-run prod         # show the database and owner each
+#                                          # launcher names, touch nothing
 #
 # Prerequisites:
 #   - psql installed and on PATH
@@ -44,8 +46,21 @@ warn()    { echo -e "${YELLOW}[⚠]${NC} $*"; }
 error()   { echo -e "${RED}[✗]${NC} $*" >&2; }
 separator(){ echo "────────────────────────────────────────────────────────────"; }
 
+# launcher_var FILE NAME — the value of `\set NAME value` in a launcher, quotes
+# stripped; empty if the launcher doesn't set it.
+launcher_var() {
+   sed -nE "s/^[[:space:]]*\\\\set[[:space:]]+$2[[:space:]]+'?([^'[:space:]]*)'?.*/\\1/p" "$1" | head -n 1
+}
+
 # ── Parse arguments ───────────────────────────────────────────────────────────
-ENVS_TO_DEPLOY=("${@:-${ALL_ENVS[@]}}")
+# --dry-run prints each environment's target database and owner, as read from
+# its launcher, and exits before touching PostgreSQL.
+DRY_RUN=false
+REQUESTED=()
+for arg in "$@"; do
+   if [[ "${arg}" == "--dry-run" ]]; then DRY_RUN=true; else REQUESTED+=("${arg}"); fi
+done
+ENVS_TO_DEPLOY=("${REQUESTED[@]:-${ALL_ENVS[@]}}")
 
 # Validate each requested environment
 for env in "${ENVS_TO_DEPLOY[@]}"; do
@@ -75,10 +90,30 @@ for env in "${ENVS_TO_DEPLOY[@]}"; do
       continue
    fi
 
-   DB_NAME="te_mgmt_${env}"
+   # The launcher is the single source of truth for the database name and owner
+   # (#172). Hard-coding them here let the deployer disagree with the launcher
+   # and with the infrastructure that built the server. The defaults below are
+   # the on-prem names, used only when a launcher doesn't set them.
+   DB_NAME="$(launcher_var "${env_file}" db_name)"
+   DB_NAME="${DB_NAME:-te_mgmt_${env}}"
+   DB_OWNER="$(launcher_var "${env_file}" db_owner)"
+   DB_OWNER="${DB_OWNER:-postgres}"
+   unsafe=""
+   for ident in "${DB_NAME}" "${DB_OWNER}"; do
+      [[ "${ident}" =~ ^[a-z_][a-z0-9_]*$ ]] || unsafe="${ident}"
+   done
+   if [[ -n "${unsafe}" ]]; then
+      error "Unsafe identifier '${unsafe}' in ${env_file} (use lower-case letters, digits and underscores)"
+      FAILED+=("${env}")
+      continue
+   fi
 
    echo ""
-   warn "Deploying environment: ${env^^}  (database: ${DB_NAME})"
+   warn "Deploying environment: ${env^^}  (database: ${DB_NAME}, owner: ${DB_OWNER})"
+   if [[ "${DRY_RUN}" == true ]]; then
+      SUCCEEDED+=("${env}")
+      continue
+   fi
 
    # Create the database if it doesn't exist.
    # CREATE DATABASE cannot run inside a transaction, so we handle it here
@@ -89,7 +124,7 @@ for env in "${ENVS_TO_DEPLOY[@]}"; do
       log "Creating database: ${DB_NAME}"
       ${PSQL} -d postgres -c \
          "CREATE DATABASE \"${DB_NAME}\"
-            WITH OWNER = postgres
+            WITH OWNER = \"${DB_OWNER}\"
             ENCODING = 'UTF8'
             TEMPLATE = template0
             CONNECTION LIMIT = -1" || {
diff --git a/build/schema/postgresql/te_core_schema.sql b/build/schema/postgresql/te_core_schema.sql
index 0573500..d73c2cc 100644
--- a/build/schema/postgresql/te_core_schema.sql
+++ b/build/schema/postgresql/te_core_schema.sql
@@ -31,12 +31,15 @@
 
 -- Store psql variables as session parameters so DO blocks can access them.
 -- (psql does not substitute :varname inside $$ dollar-quoted strings.)
+-- \gset, not ';': it keeps the result out of psql's output. The row contains the
+-- application password, and deployment output ends up in container logs (#172).
 SELECT
-   set_config('te.env_label',    :'env_label',    false),
-   set_config('te.db_name',      :'db_name',      false),
-   set_config('te.app_user',     :'app_user',     false),
-   set_config('te.app_password', :'app_password', false),
-   set_config('te.conn_limit',   :'conn_limit',   false);
+   set_config('te.env_label',    :'env_label',    false) AS env_label,
+   set_config('te.db_name',      :'db_name',      false) AS db_name,
+   set_config('te.app_user',     :'app_user',     false) AS app_user,
+   set_config('te.app_password', :'app_password', false) AS app_password,
+   set_config('te.conn_limit',   :'conn_limit',   false) AS conn_limit
+\gset te_set_
 
 \echo '>> [1/6] Creating database:' :db_name
 -- Database creation is handled by deploy_all.sh before this script runs.
@@ -615,6 +618,12 @@ BEGIN
 END;
 $$;
 
+-- The newest tenant T&E migration (alembic_te/) this schema already contains. api/tenancy.py
+-- clones this schema into each new tenant and stamps the tenant at this version, then runs
+-- any later migrations. Adding a migration means bumping this to its revision id, in both
+-- copies of this file; tests/test_tenant_te_migrations.py fails if it is left behind.
+COMMENT ON SCHEMA :"schema_name" IS 'te_schema_version=te0002';
+
 
 -- =============================================================================
 -- PHASE 5: TRIGGERS — auto-update updated_at
diff --git a/build/te_core_schema.sql b/build/te_core_schema.sql
index 0573500..d73c2cc 100644
--- a/build/te_core_schema.sql
+++ b/build/te_core_schema.sql
@@ -31,12 +31,15 @@
 
 -- Store psql variables as session parameters so DO blocks can access them.
 -- (psql does not substitute :varname inside $$ dollar-quoted strings.)
+-- \gset, not ';': it keeps the result out of psql's output. The row contains the
+-- application password, and deployment output ends up in container logs (#172).
 SELECT
-   set_config('te.env_label',    :'env_label',    false),
-   set_config('te.db_name',      :'db_name',      false),
-   set_config('te.app_user',     :'app_user',     false),
-   set_config('te.app_password', :'app_password', false),
-   set_config('te.conn_limit',   :'conn_limit',   false);
+   set_config('te.env_label',    :'env_label',    false) AS env_label,
+   set_config('te.db_name',      :'db_name',      false) AS db_name,
+   set_config('te.app_user',     :'app_user',     false) AS app_user,
+   set_config('te.app_password', :'app_password', false) AS app_password,
+   set_config('te.conn_limit',   :'conn_limit',   false) AS conn_limit
+\gset te_set_
 
 \echo '>> [1/6] Creating database:' :db_name
 -- Database creation is handled by deploy_all.sh before this script runs.
@@ -615,6 +618,12 @@ BEGIN
 END;
 $$;
 
+-- The newest tenant T&E migration (alembic_te/) this schema already contains. api/tenancy.py
+-- clones this schema into each new tenant and stamps the tenant at this version, then runs
+-- any later migrations. Adding a migration means bumping this to its revision id, in both
+-- copies of this file; tests/test_tenant_te_migrations.py fails if it is left behind.
+COMMENT ON SCHEMA :"schema_name" IS 'te_schema_version=te0002';
+
 
 -- =============================================================================
 -- PHASE 5: TRIGGERS — auto-update updated_at
diff --git a/csv-table-hub-main/src/__tests__/job-outcome.test.ts b/csv-table-hub-main/src/__tests__/job-outcome.test.ts
index 7aedf2d..162107f 100644
--- a/csv-table-hub-main/src/__tests__/job-outcome.test.ts
+++ b/csv-table-hub-main/src/__tests__/job-outcome.test.ts
@@ -8,7 +8,7 @@
 import { describe, it, expect } from "vitest";
 import {
   classifyUploadResult, classifyThrownError, buildCancelPatch,
-  buildCancelToastMessage, canStop, decideOverwriteAction, failedRowsNote,
+  buildCancelToastMessage, canStop, decideOverwriteAction, failedRowsNote, typesProvenanceNote,
 } from "@/lib/job-outcome";
 import type { UploadResult } from "@/lib/csv.functions";
 import type { Job } from "@/lib/job-types";
@@ -156,3 +156,16 @@ describe("failedRowsNote (#123)", () => {
     expect(failedRowsNote({})).toBeNull();
   });
 });
+
+describe("typesProvenanceNote (#195)", () => {
+  it("says so when the column types were never confirmed", () => {
+    expect(typesProvenanceNote({ types_provenance: "inferred_unconfirmed" })).toBe(
+      "Column types were not confirmed by a person.",
+    );
+  });
+  it("is silent for confirmed tables, T&E tables and rows from before the flag", () => {
+    expect(typesProvenanceNote({ types_provenance: "user_confirmed" })).toBeNull();
+    expect(typesProvenanceNote({ types_provenance: null })).toBeNull();
+    expect(typesProvenanceNote({})).toBeNull();
+  });
+});
diff --git a/csv-table-hub-main/src/__tests__/uploader-state-machine.test.ts b/csv-table-hub-main/src/__tests__/uploader-state-machine.test.ts
index c32e38a..e5589f5 100644
--- a/csv-table-hub-main/src/__tests__/uploader-state-machine.test.ts
+++ b/csv-table-hub-main/src/__tests__/uploader-state-machine.test.ts
@@ -208,4 +208,27 @@ describe("Uploader state machine", () => {
     expect(Array.from(Uint8Array.from(atob(call.contentBase64), (c) => c.charCodeAt(0)))).toEqual(Array.from(raw));
     expect(call.content.startsWith("id,name")).toBe(true); // the BOM is not in the decoded text
   });
+
+  it("marks the types as confirmed when they come from the user's confirmed mapping (#195)", async () => {
+    uploadCsv.mockResolvedValue({
+      status: "ok",
+      fileId: "1",
+      tableName: "csv_abc123",
+      totalRows: 2,
+      insertedRows: 2,
+      duplicateRowsSkipped: 0,
+      failedRows: 0,
+      columns: ["id", "name"],
+      types: ["int8", "text"],
+      typesProvenance: "user_confirmed",
+      rowErrors: [],
+      logs: [],
+    });
+    await dropAndConfirm(csvFile());
+
+    await waitFor(() => expect(uploadCsv).toHaveBeenCalled());
+    const call = uploadCsv.mock.calls[0][0];
+    expect(call.types).toEqual(["int8", "text"]);
+    expect(call.typesConfirmed).toBe(true);
+  });
 });
diff --git a/csv-table-hub-main/src/lib/csv.functions.ts b/csv-table-hub-main/src/lib/csv.functions.ts
index bd07654..30e2f2e 100644
--- a/csv-table-hub-main/src/lib/csv.functions.ts
+++ b/csv-table-hub-main/src/lib/csv.functions.ts
@@ -36,6 +36,8 @@ const processingLogSchema = z.object({
 });
 export type ProcessingLog = z.infer<typeof processingLogSchema>;
 
+const typesProvenanceSchema = z.enum(["inferred_unconfirmed", "user_confirmed"]);
+
 // #187: the API sends the stored attempt's id on every outcome, failures included.
 const uploadIdField = { uploadId: z.string().nullable().optional() };
 
@@ -51,6 +53,7 @@ const uploadResultSchema = z.discriminatedUnion("status", [
     failedRows: z.number(),
     columns: z.array(z.string()),
     types: z.array(columnTypeSchema),
+    typesProvenance: typesProvenanceSchema.optional(), // dynamic mode only (#195)
     rowErrors: z.array(rowErrorSchema),
     logs: z.array(processingLogSchema),
     overwritten: z.boolean().optional(),
@@ -98,6 +101,7 @@ const csvFileSummarySchema = z.object({
   failed_rows: z.number().nullable().optional(),
   uploader: z.string().nullable().optional(),
   upload_id: z.string().nullable().optional(),
+  types_provenance: typesProvenanceSchema.nullable().optional(), // null for T&E tables (#195)
   column_names: z.array(z.string()),
   created_at: z.string(),
 });
@@ -164,6 +168,7 @@ export async function uploadCsv(input: {
   targetTable?: string;
   headerMapping?: { original: string; renamed: string }[];
   contentBase64?: string;
+  typesConfirmed?: boolean;
 }): Promise<UploadResult> {
   return apiFetch("/api/csv/upload", uploadResultSchema, {
     method: "POST",
@@ -176,6 +181,7 @@ export async function uploadCsv(input: {
       targetTable: input.targetTable,
       headerMapping: input.headerMapping,
       contentBase64: input.contentBase64,
+      typesConfirmed: input.typesConfirmed,
     }),
   });
 }
diff --git a/csv-table-hub-main/src/lib/job-outcome.ts b/csv-table-hub-main/src/lib/job-outcome.ts
index af9410b..324c1a1 100644
--- a/csv-table-hub-main/src/lib/job-outcome.ts
+++ b/csv-table-hub-main/src/lib/job-outcome.ts
@@ -156,3 +156,13 @@ export function failedRowsNote(f: {
     ? `${f.failed_rows} of ${f.total_rows} rows failed`
     : `${f.failed_rows} rows failed`;
 }
+
+/**
+ * Says so when a dynamic table's column types were never confirmed by a person (#195);
+ * null for confirmed tables and for T&E tables, which have no inferred types.
+ */
+export function typesProvenanceNote(f: { types_provenance?: string | null }): string | null {
+  return f.types_provenance === "inferred_unconfirmed"
+    ? "Column types were not confirmed by a person."
+    : null;
+}
diff --git a/csv-table-hub-main/src/routes/_authenticated/index.tsx b/csv-table-hub-main/src/routes/_authenticated/index.tsx
index 75a330c..2233d29 100644
--- a/csv-table-hub-main/src/routes/_authenticated/index.tsx
+++ b/csv-table-hub-main/src/routes/_authenticated/index.tsx
@@ -38,7 +38,7 @@ import { readFileBase64 } from "@/lib/file-bytes";
 import { formatLocalTime, formatUtc } from "@/lib/utils";
 import {
   classifyUploadResult, classifyThrownError, buildCancelPatch, buildCancelToastMessage, canStop,
-  decideOverwriteAction, failedRowsNote,
+  decideOverwriteAction, failedRowsNote, typesProvenanceNote,
 } from "@/lib/job-outcome";
 import { needsAttention } from "@/lib/job-outcome";
 import { Button } from "@/components/ui/button";
@@ -188,6 +188,7 @@ export function Uploader() {
 
       const res = await uploadCsv({
         fileName:job.name, content, contentBase64, types, overwrite,
+        typesConfirmed: !!types?.length, // #195: the types come from the mapping the user confirmed
         headerMapping: headerMapping?.map(c=>({original:c.original,renamed:c.renamed})),
         mode: job.mode, targetTable: job.targetTable,
       });
@@ -781,7 +782,7 @@ function PreviewTable({file}:{file:CsvFileSummary}){
           {rows.length===0&&<TableRow><TableCell colSpan={file.column_names.length} className="text-center text-muted-foreground">No rows.</TableCell></TableRow>}
         </TableBody>
       </Table>
-      <p className="p-2 text-xs text-muted-foreground">Showing up to 50 rows. Table has {file.row_count} row{file.row_count===1?"":"s"} total.{failedRowsNote(file)&&<span className="text-destructive"> {failedRowsNote(file)} validation and weren't imported.</span>}</p>
+      <p className="p-2 text-xs text-muted-foreground">Showing up to 50 rows. Table has {file.row_count} row{file.row_count===1?"":"s"} total.{failedRowsNote(file)&&<span className="text-destructive"> {failedRowsNote(file)} validation and weren't imported.</span>}{typesProvenanceNote(file)&&<span className="text-amber-600 dark:text-amber-500"> {typesProvenanceNote(file)}</span>}</p>
       {file.upload_id&&(file.failed_rows??0)>0&&<StoredRowErrors uploadId={file.upload_id} fileName={file.file_name}/>}
     </div>
   );
diff --git a/docs/DATABASE_ENVIRONMENTS.md b/docs/DATABASE_ENVIRONMENTS.md
index 4a53816..5cebb2c 100644
--- a/docs/DATABASE_ENVIRONMENTS.md
+++ b/docs/DATABASE_ENVIRONMENTS.md
@@ -16,6 +16,14 @@ for names: `build/config.env.example` (template) / `build/config.local.env`
 | **staging** | `te_mgmt_staging` | `te_staging` | `te_stg_user` | 25 | no (`SEED_STAGING=false`) |
 | **prod** | `te_mgmt_prod` | `te_prod` | `te_prod_user` | 50 | no (`SEED_PROD=false`) |
 
+The database name and owner are read from each environment's launcher
+(`build/environments/env_<env>.sql`, `db_name` and `db_owner`) by
+`build/deploy_all.sh`; the table above is what the committed templates set.
+**Azure production differs:** the database is `te_prod`, owned by `pgadmin`, because
+Terraform creates it and Azure has no `postgres` role. The migration job's
+entrypoint writes that into its launcher at run time. See
+`DECISIONS/2026-11-09-prod-db-naming.md`.
+
 Every environment also gets a `csv_uploads` schema in its own database
 (same name across environments — no collision, since each environment is a
 separate database). Controlled by `CSV_UPLOADS_SCHEMA` (default
diff --git a/docs/TENANT_SCHEMA_MIGRATIONS.md b/docs/TENANT_SCHEMA_MIGRATIONS.md
new file mode 100644
index 0000000..8f400f8
--- /dev/null
+++ b/docs/TENANT_SCHEMA_MIGRATIONS.md
@@ -0,0 +1,73 @@
+# Changing the T&E tables for tenants
+
+Each tenant has its own copy of the twelve T&E tables in schema `tenant_<slug>`
+(`api/tenancy.py`). The deployment's template schema (`te_dev`, `te_prod`, ...) is
+built by `build/te_core_schema.sql`. A tenant's copy is made from the template once;
+after that, **a change to the template does not reach it by itself**. Tenant T&E
+migrations (#184) are what carry a change across.
+
+## How a tenant's T&E schema is kept current
+
+`migrate_tenant(slug)` runs at API start for every tenant, and from
+`scripts/create_tenant_admin.py`. For the T&E tables it does one of three things,
+serialised per tenant by an advisory lock:
+
+| Tenant state | What happens |
+|---|---|
+| No T&E tables | Cloned from the template, stamped at the template's version, then upgraded to head |
+| Tables, no version table (made before this existed) | Stamped at the baseline `te0001`, then upgraded to head |
+| Version table present | Upgraded to head |
+
+The version lives in `alembic_version_te`, inside the tenant schema. The migrations
+are in `alembic_te/versions/`; they are a separate chain from `alembic/`, which
+versions the uploads objects (`csv_files`, `audit_log`, ...).
+
+An upgrade is one transaction. If it fails, that tenant is left exactly as it was and
+the error says why. `te0002` (program consistency, #177) refuses to run if a tenant
+already has rows that join two programs, and lists up to 50 of them: fix or remove
+those rows, then restart.
+
+## Making a change to the T&E tables
+
+Every change is made in three places, and a test fails if one is missed:
+
+1. **Template**: edit `build/te_core_schema.sql` **and** its identical copy
+   `build/schema/postgresql/te_core_schema.sql`. For existing databases, make the
+   change idempotent (see phase 4B for the pattern).
+2. **Migration**: add `alembic_te/versions/te000N_<name>.py` (revises the previous
+   head) doing the same to a tenant schema. Read the schema with
+   `op.get_context().opts["version_table_schema"]`. Make `upgrade()` idempotent, and
+   write a `downgrade()`.
+3. **Stamp**: change the `COMMENT ON SCHEMA ... IS 'te_schema_version=...'` line in
+   both schema files to the new revision id.
+
+`tests/system/test_tenant_te_migrations.py` then checks that a tenant reached by any route
+(cloned, upgraded from the baseline, adopted from before versioning, cloned from an
+older template) has the same columns, constraints, indexes, triggers and functions as
+the template, and that both schema files carry the head's stamp. A change made in
+the template only, or in the migration only, fails the comparison.
+
+Index and unique-constraint names matter: provisioning renames the cloned ones to the
+template's names, so a migration can drop or rename them by name.
+
+## Deploying
+
+- **New template, old tenants.** Deploy the API; tenants are upgraded at start. The
+  template schema does not have to be redeployed first, because a tenant upgrade does
+  not read it.
+- **New tenants.** The template must carry a version stamp. A template deployed by an
+  older `te_core_schema.sql` has none, and provisioning a tenant says so and asks you
+  to redeploy it with `build/deploy_all.sh`. A template *behind* the API is fine: the
+  new tenant is cloned, stamped at the template's version, and brought forward. A
+  template *ahead* of the API (stamped with a revision this release doesn't know) is
+  refused.
+- **Rolling back.** `alembic_te` supports `downgrade`; tests use it to build the
+  pre-#177 shape. It is not run automatically.
+
+## Known limits
+
+- Tenants cloned by provisioning before #184 keep PostgreSQL's default **index** names
+  (for example `test_cases_phase_id_idx`) on indexes the migrations don't touch. Their
+  unique constraints are renamed by `te0002`. New tenants have the template's names.
+- The migration SQL and `te_core_schema.sql` are two copies of the same change. The
+  comparison test is what keeps them agreeing; there is no generator.
diff --git a/docs/TEST_CASES.md b/docs/TEST_CASES.md
index 314f585..2d4ad3d 100644
--- a/docs/TEST_CASES.md
+++ b/docs/TEST_CASES.md
@@ -1,7 +1,7 @@
 # Test Catalogue
 
 > Auto-generated by `scripts/test_catalogue.py` — do not edit by hand.
-> 809 tests across 40 files.
+> 897 tests across 43 files.
 
 To regenerate: `python3 scripts/test_catalogue.py`
 
@@ -567,6 +567,12 @@ To regenerate: `python3 scripts/test_catalogue.py`
 
 | `MigrationsOnScratchSchema.test_baseline_adopts_a_fallback_created_schema` | integration | — |
 
+| `MigrationsOnScratchSchema.test_type_provenance_marks_old_dynamic_rows_unconfirmed_and_leaves_te_null` | integration | — |
+
+| `MigrationsOnScratchSchema.test_type_provenance_only_accepts_the_two_known_values` | integration | — |
+
+| `MigrationsOnScratchSchema.test_type_provenance_downgrade_drops_the_column` | integration | — |
+
 | `MigrationsOnScratchSchema.test_existing_duplicates_stop_the_upgrade_with_remediation` | integration | — |
 
 | `RegistryUniqueness.test_bootstrap_brings_the_schema_to_head` | integration | — |
@@ -585,6 +591,12 @@ To regenerate: `python3 scripts/test_catalogue.py`
 
 | `test_baseline_adopts_a_fallback_created_schema` | — | — |
 
+| `test_type_provenance_marks_old_dynamic_rows_unconfirmed_and_leaves_te_null` | — | — |
+
+| `test_type_provenance_only_accepts_the_two_known_values` | — | — |
+
+| `test_type_provenance_downgrade_drops_the_column` | — | — |
+
 | `test_existing_duplicates_stop_the_upgrade_with_remediation` | — | — |
 
 | `test_bootstrap_brings_the_schema_to_head` | — | — |
@@ -1051,6 +1063,110 @@ To regenerate: `python3 scripts/test_catalogue.py`
 
 | `test_provisioning_is_idempotent` | — | — |
 
+## `tests/system/test_tenant_te_migrations.py`
+
+| Test | Markers | Objective |
+|---|---|---|
+| `TheChain.test_it_is_linear_and_starts_at_the_baseline` | unit | — |
+
+| `TheChain.test_both_copies_of_the_schema_file_stamp_the_head` | unit | — |
+
+| `TheChain.test_the_runtime_role_cannot_rewrite_the_version_table` | unit | — |
+
+| `TenantTeSchemaUpgrades.test_a_new_tenant_is_cloned_stamped_at_head_and_matches_the_template` | integration | — |
+
+| `TenantTeSchemaUpgrades.test_the_template_is_stamped_with_the_head` | integration | — |
+
+| `TenantTeSchemaUpgrades.test_running_again_changes_nothing` | integration | — |
+
+| `TenantTeSchemaUpgrades.test_downgrade_restores_the_shape_from_before_177_and_upgrade_restores_the_template` | integration | — |
+
+| `TenantTeSchemaUpgrades.test_a_tenant_from_before_versioning_is_adopted_and_keeps_its_data` | integration | — |
+
+| `TenantTeSchemaUpgrades.test_a_tenant_cloned_after_177_with_default_constraint_names_is_adopted_without_duplicates` | integration | — |
+
+| `TenantTeSchemaUpgrades.test_a_clone_carries_the_templates_index_and_constraint_names` | integration | — |
+
+| `TenantTeSchemaUpgrades.test_cross_program_rows_stop_the_upgrade_name_the_rows_and_change_nothing` | integration | — |
+
+| `TenantTeSchemaUpgrades.test_a_template_at_an_older_version_is_cloned_then_brought_forward` | integration | — |
+
+| `TenantTeSchemaUpgrades.test_the_next_schema_change_reaches_existing_and_new_tenants` | integration | The reproduction in #184: change the T&E tables after a tenant exists and it must get the change. |
+
+| `TenantTeSchemaUpgrades.test_a_template_with_no_stamp_is_refused_with_the_way_out` | integration | — |
+
+| `TenantTeSchemaUpgrades.test_a_template_stamped_with_an_unknown_version_is_refused` | integration | — |
+
+| `TenantTeSchemaUpgrades.test_two_workers_starting_together_leave_one_correct_tenant` | integration | — |
+
+| `test_it_is_linear_and_starts_at_the_baseline` | — | — |
+
+| `test_both_copies_of_the_schema_file_stamp_the_head` | — | — |
+
+| `test_the_runtime_role_cannot_rewrite_the_version_table` | — | — |
+
+| `test_a_new_tenant_is_cloned_stamped_at_head_and_matches_the_template` | — | — |
+
+| `test_the_template_is_stamped_with_the_head` | — | — |
+
+| `test_running_again_changes_nothing` | — | — |
+
+| `test_downgrade_restores_the_shape_from_before_177_and_upgrade_restores_the_template` | — | — |
+
+| `test_a_tenant_from_before_versioning_is_adopted_and_keeps_its_data` | — | — |
+
+| `test_a_tenant_cloned_after_177_with_default_constraint_names_is_adopted_without_duplicates` | — | — |
+
+| `test_a_clone_carries_the_templates_index_and_constraint_names` | — | — |
+
+| `test_cross_program_rows_stop_the_upgrade_name_the_rows_and_change_nothing` | — | — |
+
+| `test_a_template_at_an_older_version_is_cloned_then_brought_forward` | — | — |
+
+| `test_the_next_schema_change_reaches_existing_and_new_tenants` | — | The reproduction in #184: change the T&E tables after a tenant exists and it must get the change. |
+
+| `test_a_template_with_no_stamp_is_refused_with_the_way_out` | — | — |
+
+| `test_a_template_stamped_with_an_unknown_version_is_refused` | — | — |
+
+| `test_two_workers_starting_together_leave_one_correct_tenant` | — | — |
+
+## `tests/system/test_type_provenance.py`
+
+| Test | Markers | Objective |
+|---|---|---|
+| `TypeConfirmationAndProvenance.test_preview_proposes_types_but_creates_no_table_and_no_registry_row` | integration | — |
+
+| `TypeConfirmationAndProvenance.test_confirming_creates_the_table_with_the_confirmed_types` | integration | — |
+
+| `TypeConfirmationAndProvenance.test_the_confirmed_types_can_differ_from_what_was_inferred` | integration | — |
+
+| `TypeConfirmationAndProvenance.test_types_sent_without_confirmation_are_applied_but_not_marked_confirmed` | integration | — |
+
+| `TypeConfirmationAndProvenance.test_no_types_at_all_is_unconfirmed` | integration | — |
+
+| `TypeConfirmationAndProvenance.test_confirmation_without_the_confirmed_types_is_rejected_and_loads_nothing` | integration | — |
+
+| `TypeConfirmationAndProvenance.test_te_mode_rejects_types_confirmed` | integration | — |
+
+| `TypeConfirmationAndProvenance.test_te_import_rejects_a_dynamic_table_as_its_target` | integration | — |
+
+| `test_preview_proposes_types_but_creates_no_table_and_no_registry_row` | — | — |
+
+| `test_confirming_creates_the_table_with_the_confirmed_types` | — | — |
+
+| `test_the_confirmed_types_can_differ_from_what_was_inferred` | — | — |
+
+| `test_types_sent_without_confirmation_are_applied_but_not_marked_confirmed` | — | — |
+
+| `test_no_types_at_all_is_unconfirmed` | — | — |
+
+| `test_confirmation_without_the_confirmed_types_is_rejected_and_loads_nothing` | — | — |
+
+| `test_te_mode_rejects_types_confirmed` | — | — |
+
+| `test_te_import_rejects_a_dynamic_table_as_its_target` | — | — |
+
 ## `tests/system_integration/test_csv_loader_arbitrary_shapes.py`
 
 | Test | Markers | Objective |
@@ -1683,6 +1799,78 @@ To regenerate: `python3 scripts/test_catalogue.py`
 
 | `test_pg_checks_are_optional_never_fail` | regression | — |
 
+## `tests/unit/test_prod_db_naming.py`
+
+| Test | Markers | Objective |
+|---|---|---|
+| `DeployerReadsNamesFromTheLauncher.test_committed_templates_keep_the_on_prem_names` | unit | — |
+
+| `DeployerReadsNamesFromTheLauncher.test_the_launcher_decides_the_database_and_owner` | unit | — |
+
+| `DeployerReadsNamesFromTheLauncher.test_a_launcher_that_sets_neither_falls_back_to_the_on_prem_names` | unit | — |
+
+| `DeployerReadsNamesFromTheLauncher.test_an_unsafe_identifier_is_refused` | unit | — |
+
+| `EntrypointWritesTheLauncher.test_the_image_has_no_concrete_launcher` | unit | — |
+
+| `EntrypointWritesTheLauncher.test_the_launcher_is_made_from_the_containers_database_and_user` | unit | — |
+
+| `EntrypointWritesTheLauncher.test_the_password_is_read_by_psql_and_never_written` | unit | — |
+
+| `EntrypointWritesTheLauncher.test_a_missing_password_is_an_error` | unit | — |
+
+| `EntrypointWritesTheLauncher.test_an_unsafe_database_name_is_an_error` | unit | — |
+
+| `EntrypointWritesTheLauncher.test_an_existing_launcher_is_left_alone` | unit | — |
+
+| `EntrypointWritesTheLauncher.test_an_environment_with_no_template_is_an_error` | unit | — |
+
+| `TerraformAgreesWithTheDeployment.test_the_database_terraform_creates_is_te_prod` | unit | — |
+
+| `TerraformAgreesWithTheDeployment.test_the_job_deploys_into_that_database_as_the_servers_admin` | unit | — |
+
+| `TerraformAgreesWithTheDeployment.test_the_job_receives_the_application_password_as_a_secret` | unit | — |
+
+| `TerraformAgreesWithTheDeployment.test_every_extension_the_schema_creates_is_allow_listed_on_azure` | unit | — |
+
+| `DeploymentOutputDoesNotEchoThePassword.test_the_two_schema_copies_are_identical` | unit | — |
+
+| `DeploymentOutputDoesNotEchoThePassword.test_the_password_row_is_captured_not_printed` | unit | — |
+
+| `test_committed_templates_keep_the_on_prem_names` | — | — |
+
+| `test_the_launcher_decides_the_database_and_owner` | — | — |
+
+| `test_a_launcher_that_sets_neither_falls_back_to_the_on_prem_names` | — | — |
+
+| `test_an_unsafe_identifier_is_refused` | — | — |
+
+| `test_the_image_has_no_concrete_launcher` | — | — |
+
+| `test_the_launcher_is_made_from_the_containers_database_and_user` | — | — |
+
+| `test_the_password_is_read_by_psql_and_never_written` | — | — |
+
+| `test_a_missing_password_is_an_error` | — | — |
+
+| `test_an_unsafe_database_name_is_an_error` | — | — |
+
+| `test_an_existing_launcher_is_left_alone` | — | — |
+
+| `test_an_environment_with_no_template_is_an_error` | — | — |
+
+| `test_the_database_terraform_creates_is_te_prod` | — | — |
+
+| `test_the_job_deploys_into_that_database_as_the_servers_admin` | — | — |
+
+| `test_the_job_receives_the_application_password_as_a_secret` | — | — |
+
+| `test_every_extension_the_schema_creates_is_allow_listed_on_azure` | — | — |
+
+| `test_the_two_schema_copies_are_identical` | — | — |
+
+| `test_the_password_row_is_captured_not_printed` | — | — |
+
 ## `tests/unit/test_regression.py`
 
 | Test | Markers | Objective |
diff --git a/infra/Dockerfile b/infra/Dockerfile
index 91893c9..4d72d81 100644
--- a/infra/Dockerfile
+++ b/infra/Dockerfile
@@ -5,7 +5,16 @@
 #   2. Load input CSVs (input_data/load_input_data.sql)
 #   3. Run the eval suite (evals/runner.py --tiers p,i,s)
 #
-# Entrypoint dispatches on the first arg: deploy | load | evals | shell.
+# Entrypoint dispatches on the first arg: deploy | render | load | evals | full | shell.
+#
+# The image holds only the launcher TEMPLATES (build/environments/*.example.sql);
+# the concrete env_<env>.sql carries credentials and is never baked in (#174).
+# `deploy` writes it at run time from the container's settings (#172):
+#   PGDATABASE   -> the database to deploy into (db_name)
+#   PGUSER       -> its owner and the login that deploys (db_owner)
+#   APP_DB_PASSWORD -> password of the application role the deployment creates;
+#                  passed to psql by \getenv, never written to a file or logged.
+# `render` prints the launcher it would use, without touching PostgreSQL.
 #
 # Build from the PostgreDataMigrationApp repo root:
 #   docker build -f infra/Dockerfile -t te-migration:dev .
@@ -17,6 +26,7 @@
 #   docker run --rm \
 #     -e PGHOST=postgres -e PGPORT=5432 -e PGUSER=postgres \
 #     -e PGPASSWORD=devpass -e PGDATABASE=te_dev \
+#     -e APP_DB_PASSWORD=app-devpass \
 #     -e TARGET_ENV=dev \
 #     te-migration:dev deploy
 
diff --git a/infra/entrypoint.sh b/infra/entrypoint.sh
index e8a6c43..28e800c 100644
--- a/infra/entrypoint.sh
+++ b/infra/entrypoint.sh
@@ -1,10 +1,11 @@
 #!/usr/bin/env bash
 # entrypoint.sh - dispatches to the correct workflow inside the container.
-# Usage (passed as the container CMD): deploy | load | evals | full | shell
+# Usage (passed as the container CMD): deploy | load | evals | full | render | shell
 
 set -euo pipefail
 
 ACTION="${1:-deploy}"
+MIGRATION_HOME="${MIGRATION_HOME:-/opt/migration}"
 
 echo "============================================================"
 echo " T&E Migration container"
@@ -30,16 +31,66 @@ wait_for_pg() {
     echo "PostgreSQL is ready."
 }
 
+# The image carries only the committed launcher templates
+# (build/environments/env_<env>.example.sql); the concrete env_<env>.sql holds
+# credentials, is gitignored, and must never be baked into an image (#174). So
+# make it here, at run time, from this container's own settings (#172):
+#   db_name     <- PGDATABASE   the database the infrastructure created
+#   db_owner    <- PGUSER       the login that runs the deployment
+#   app_password <- APP_DB_PASSWORD, read by psql (\getenv) when the deployment
+#                  runs, so the secret is never written to disk.
+# An existing launcher (mounted in, or from a developer's checkout) is left alone.
+ensure_launcher() {
+    local dir="${MIGRATION_HOME}/build/environments"
+    local launcher="${dir}/env_${TARGET_ENV}.sql"
+    local template="${dir}/env_${TARGET_ENV}.example.sql"
+
+    if [ -f "$launcher" ]; then
+        echo "--- Using existing launcher: ${launcher} ---"
+        return 0
+    fi
+    if [ ! -f "$template" ]; then
+        echo "ERROR: no launcher for TARGET_ENV='${TARGET_ENV}': neither ${launcher}" >&2
+        echo "       nor its template ${template} exists." >&2
+        exit 1
+    fi
+    if [ -z "${APP_DB_PASSWORD:-}" ]; then
+        echo "ERROR: APP_DB_PASSWORD is not set. It is the password for the application" >&2
+        echo "       role this deployment creates; supply it as a secret environment variable." >&2
+        exit 1
+    fi
+    local ident
+    for ident in "$PGDATABASE" "$PGUSER"; do
+        if [[ ! "$ident" =~ ^[a-z_][a-z0-9_]*$ ]]; then
+            echo "ERROR: '${ident}' (PGDATABASE / PGUSER) must be lower-case letters, digits and underscores." >&2
+            exit 1
+        fi
+    done
+
+    sed -E \
+        -e "s/^([[:space:]]*\\\\set[[:space:]]+db_name[[:space:]]+).*/\\1${PGDATABASE}/" \
+        -e "s/^([[:space:]]*\\\\set[[:space:]]+db_owner[[:space:]]+).*/\\1${PGUSER}/" \
+        -e "s/^[[:space:]]*\\\\set[[:space:]]+app_password[[:space:]].*/\\\\getenv app_password APP_DB_PASSWORD/" \
+        "$template" > "$launcher"
+    echo "--- Generated ${launcher} from its template (db ${PGDATABASE}, owner ${PGUSER}) ---"
+}
+
 run_deploy() {
     wait_for_pg
+    ensure_launcher
     echo "--- Deploying schema for env: ${TARGET_ENV} ---"
-    cd /opt/migration
+    cd "$MIGRATION_HOME"
     bash build/deploy_all.sh "${TARGET_ENV}"
 }
 
+run_render() {
+    ensure_launcher
+    cat "${MIGRATION_HOME}/build/environments/env_${TARGET_ENV}.sql"
+}
+
 run_load() {
     wait_for_pg
-    if [ ! -f /opt/migration/input_data/load_input_data.sql ]; then
+    if [ ! -f "${MIGRATION_HOME}/input_data/load_input_data.sql" ]; then
         echo "--- SKIP load: no input_data/load_input_data.sql in image ---"
         echo "    (mount input_data as a volume or rebuild the image with it included)"
         return 0
@@ -47,13 +98,13 @@ run_load() {
     echo "--- Loading input data ---"
     psql -h "$PGHOST" -p "$PGPORT" -U "$PGUSER" -d "$PGDATABASE" \
          -v ON_ERROR_STOP=1 \
-         -f /opt/migration/input_data/load_input_data.sql
+         -f "${MIGRATION_HOME}/input_data/load_input_data.sql"
 }
 
 run_evals() {
     wait_for_pg
     echo "--- Running eval suite (tiers p,i,s) ---"
-    cd /opt/migration
+    cd "$MIGRATION_HOME"
     python3 evals/runner.py --tiers p,i,s
 }
 
@@ -65,13 +116,14 @@ run_full() {
 
 case "$ACTION" in
     deploy) run_deploy ;;
+    render) run_render ;;
     load)   run_load ;;
     evals)  run_evals ;;
     full)   run_full ;;
     shell)  exec /bin/bash ;;
     *)
         echo "Unknown action: $ACTION" >&2
-        echo "Valid: deploy | load | evals | full | shell" >&2
+        echo "Valid: deploy | render | load | evals | full | shell" >&2
         exit 2
         ;;
 esac
diff --git a/infra/terraform-prod/main.tf b/infra/terraform-prod/main.tf
index 427fe5b..74000a1 100644
--- a/infra/terraform-prod/main.tf
+++ b/infra/terraform-prod/main.tf
@@ -227,6 +227,26 @@ resource "azurerm_key_vault_secret" "pg_password" {
   depends_on = [azurerm_role_assignment.kv_admin_terraform]
 }
 
+# Password for the application role (te_prod_user) that the migration job
+# creates. The job reads it as APP_DB_PASSWORD; it is never baked into the image
+# or written to a launcher file (#172, #174).
+resource "random_password" "app_password" {
+  length           = 32
+  special          = true
+  min_lower        = 4
+  min_upper        = 4
+  min_numeric      = 4
+  override_special = "!@#$%^&*()-_=+"
+}
+
+resource "azurerm_key_vault_secret" "app_password" {
+  name         = "pg-app-password"
+  value        = random_password.app_password.result
+  key_vault_id = azurerm_key_vault.kv.id
+
+  depends_on = [azurerm_role_assignment.kv_admin_terraform]
+}
+
 # --- PostgreSQL Flexible Server (managed, HA-capable) -------------------
 
 resource "azurerm_postgresql_flexible_server" "pg" {
@@ -271,6 +291,11 @@ resource "azurerm_postgresql_flexible_server" "pg" {
   }
 }
 
+# The production database on Azure is te_prod, owned by the server's admin login
+# (pgadmin). The migration job receives both as PGDATABASE and PGUSER, and its
+# entrypoint writes them into the launcher it deploys with (infra/entrypoint.sh),
+# so this resource is the only place the name is set. The on-prem installer keeps
+# its own name, te_mgmt_prod (DECISIONS/2026-11-09-prod-db-naming.md).
 resource "azurerm_postgresql_flexible_server_database" "te_prod" {
   name      = "te_prod"
   server_id = azurerm_postgresql_flexible_server.pg.id
@@ -278,6 +303,15 @@ resource "azurerm_postgresql_flexible_server_database" "te_prod" {
   collation = "en_US.utf8"
 }
 
+# The schema needs uuid-ossp, pg_trgm and dblink. Azure refuses CREATE EXTENSION
+# for anything not allow-listed here, so without this the first deployment fails
+# partway through creating the schema.
+resource "azurerm_postgresql_flexible_server_configuration" "extensions" {
+  name      = "azure.extensions"
+  server_id = azurerm_postgresql_flexible_server.pg.id
+  value     = "dblink,pg_trgm,uuid-ossp"
+}
+
 resource "azurerm_postgresql_flexible_server_configuration" "log_statement" {
   name      = "log_statement"
   server_id = azurerm_postgresql_flexible_server.pg.id
@@ -477,6 +511,10 @@ resource "azurerm_container_app_job" "migration" {
         name        = "PGPASSWORD"
         secret_name = "pg-password"
       }
+      env {
+        name        = "APP_DB_PASSWORD"
+        secret_name = "app-password"
+      }
     }
   }
 
@@ -486,6 +524,12 @@ resource "azurerm_container_app_job" "migration" {
     key_vault_secret_id = azurerm_key_vault_secret.pg_password.id
   }
 
+  secret {
+    name                = "app-password"
+    identity            = azurerm_user_assigned_identity.runner.id
+    key_vault_secret_id = azurerm_key_vault_secret.app_password.id
+  }
+
   tags = local.tags
 
   lifecycle {
diff --git a/tests/system/test_migrations.py b/tests/system/test_migrations.py
index d372384..8f248d5 100644
--- a/tests/system/test_migrations.py
+++ b/tests/system/test_migrations.py
@@ -18,6 +18,7 @@
 import pytest
 from alembic import command
 from alembic.config import Config
+from alembic.script import ScriptDirectory
 from fastapi.testclient import TestClient
 
 from api.config import settings
@@ -114,6 +115,39 @@ def test_baseline_adopts_a_fallback_created_schema(self):
         command.upgrade(self.cfg, "head")
         self.assertIn("csv_files_file_hash_key", self._constraints())
 
+    def _provenance_by_file(self) -> list[tuple]:
+        with self.conn.cursor() as cur:
+            cur.execute(f'SELECT file_name, types_provenance FROM "{settings.UPLOADS_SCHEMA}".csv_files '
+                        "ORDER BY file_name")
+            return cur.fetchall()
+
+    def test_type_provenance_marks_old_dynamic_rows_unconfirmed_and_leaves_te_null(self):
+        command.upgrade(self.cfg, "0008")
+        insert = (f'INSERT INTO "{settings.UPLOADS_SCHEMA}".csv_files (file_name, file_hash, table_name, mode) '
+                  "VALUES (%s, %s, 't', %s)")
+        with self.conn.cursor() as cur:
+            cur.execute(insert, ("d.csv", "h_dyn", "dynamic"))
+            cur.execute(insert, ("t.csv", "h_te", "te"))
+        command.upgrade(self.cfg, "head")
+        self.assertEqual(self._provenance_by_file(), [("d.csv", "inferred_unconfirmed"), ("t.csv", None)])
+
+    def test_type_provenance_only_accepts_the_two_known_values(self):
+        command.upgrade(self.cfg, "head")
+        with self.conn.cursor() as cur:
+            with self.assertRaises(psycopg2.errors.CheckViolation):
+                cur.execute(f'INSERT INTO "{settings.UPLOADS_SCHEMA}".csv_files '
+                            "(file_name, file_hash, table_name, mode, types_provenance) "
+                            "VALUES ('x.csv', 'hx', 't', 'dynamic', 'bogus')")
+
+    def test_type_provenance_downgrade_drops_the_column(self):
+        command.upgrade(self.cfg, "head")
+        command.downgrade(self.cfg, "0008")
+        with self.conn.cursor() as cur:
+            cur.execute("SELECT 1 FROM information_schema.columns WHERE table_schema = %s "
+                        "AND table_name = 'csv_files' AND column_name = 'types_provenance'",
+                        (settings.UPLOADS_SCHEMA,))
+            self.assertIsNone(cur.fetchone())
+
     def test_existing_duplicates_stop_the_upgrade_with_remediation(self):
         command.upgrade(self.cfg, "0001")
         with self.conn.cursor() as cur:
@@ -148,7 +182,7 @@ def setUp(self):
     def test_bootstrap_brings_the_schema_to_head(self):
         with Conn() as conn, conn.cursor() as cur:
             cur.execute(f'SELECT version_num FROM "{settings.UPLOADS_SCHEMA}".alembic_version')
-            self.assertEqual(cur.fetchone()[0], "0008")
+            self.assertEqual(cur.fetchone()[0], ScriptDirectory.from_config(Config(_ALEMBIC_INI)).get_current_head())
 
     def test_duplicate_hash_is_rejected_by_the_database(self):
         insert = (f'INSERT INTO "{settings.UPLOADS_SCHEMA}".csv_files (file_name, file_hash, table_name, mode) '
diff --git a/tests/system/test_tenant_isolation.py b/tests/system/test_tenant_isolation.py
index 5fc4599..d2df5b5 100644
--- a/tests/system/test_tenant_isolation.py
+++ b/tests/system/test_tenant_isolation.py
@@ -10,9 +10,12 @@
 import os
 import unittest
 import uuid
+from pathlib import Path
 
 import psycopg2.errors
 import pytest
+from alembic.config import Config
+from alembic.script import ScriptDirectory
 from fastapi.testclient import TestClient
 from psycopg2 import sql
 
@@ -25,6 +28,7 @@
 _HELP = ("Start PostgreSQL and ensure api/ can reach it (PGHOST/PGPORT/PGUSER/"
          "PGPASSWORD/PGDATABASE). See scripts/start-api.ps1.")
 _PW = "correct horse battery staple"
+_ALEMBIC_INI = str(Path(__file__).resolve().parents[2] / "alembic.ini")
 
 
 @pytest.mark.integration
@@ -199,7 +203,7 @@ def test_tenant_schema_is_at_head_with_its_own_registry(self):
         with Conn() as conn, conn.cursor() as cur:
             cur.execute(sql.SQL("SELECT version_num FROM {}.alembic_version").format(
                 sql.Identifier(schema_for(self.slugs["b"]))))
-            self.assertEqual(cur.fetchone()[0], "0008")
+            self.assertEqual(cur.fetchone()[0], ScriptDirectory.from_config(Config(_ALEMBIC_INI)).get_current_head())
             cur.execute("SELECT table_name FROM information_schema.tables WHERE table_schema = %s",
                         (schema_for(self.slugs["b"]),))
             self.assertTrue({"csv_files", "audit_log"} <= {r[0] for r in cur.fetchall()})
diff --git a/tests/system/test_tenant_te_migrations.py b/tests/system/test_tenant_te_migrations.py
new file mode 100644
index 0000000..0997c4a
--- /dev/null
+++ b/tests/system/test_tenant_te_migrations.py
@@ -0,0 +1,346 @@
+"""Tenant T&E schemas have an upgrade path (#184).
+
+Before: api/tenancy.py cloned only the T&E tables a tenant was missing and returned. A change to
+the template (#177's composite keys) never reached a tenant that already had its tables.
+
+Now each tenant's T&E tables carry a version (alembic_te/, table alembic_version_te) and
+migrate_tenant brings them to head. The property that matters is *parity*: a tenant reached by
+any route (cloned, upgraded from the baseline, adopted from before versioning, cloned from an
+older template) ends up structurally identical to the template.
+
+The "old shape" fixture is the template's own tables run back to te0001 by the chain's downgrade,
+and its shape is asserted below, so the test does not take the downgrade on trust.
+
+Integration tests need a live PostgreSQL with the template (te_dev) deployed and seeded.
+"""
+
+from __future__ import annotations
+
+import re
+import shutil
+import tempfile
+import unittest
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from unittest import mock
+
+import pytest
+from alembic import command
+from alembic.script import ScriptDirectory
+from psycopg2 import sql
+
+from api.config import TE_TABLES, settings
+from api.db import APPEND_ONLY, owner_connect
+from api.tenancy import (TE_BASELINE, TE_VERSION_TABLE, _te_config, migrate_te_schema, migrate_tenant,
+                         provision_te_tables, schema_for, te_head, template_te_version)
+
+ROOT = Path(__file__).resolve().parents[2]
+_PROGRAM_TABLES = ("test_cases", "test_events", "vcrm_entries", "test_results")
+
+
+@pytest.mark.unit
+class TheChain(unittest.TestCase):
+    def test_it_is_linear_and_starts_at_the_baseline(self):
+        script = ScriptDirectory.from_config(_te_config("unused"))
+        self.assertEqual([r.revision for r in script.walk_revisions()], ["te0002", "te0001"])
+        self.assertEqual(TE_BASELINE, "te0001")
+
+    def test_both_copies_of_the_schema_file_stamp_the_head(self):
+        for path in ("build/te_core_schema.sql", "build/schema/postgresql/te_core_schema.sql"):
+            text = (ROOT / path).read_text(encoding="utf-8")
+            self.assertIn(f"IS 'te_schema_version={te_head()}'", text,
+                          f"{path} must stamp the newest migration; bump it with the migration")
+
+    def test_the_runtime_role_cannot_rewrite_the_version_table(self):
+        self.assertIn(TE_VERSION_TABLE, APPEND_ONLY)
+
+
+def signature(cur, schema: str) -> dict[str, list]:
+    """Everything about a schema's T&E tables that a migration could change, with its own name removed."""
+    def norm(text):
+        return None if text is None else re.sub(rf'(?<![\w"]){re.escape(schema)}\.', "<S>.", text)
+
+    sig: dict[str, list] = {}
+    cur.execute("SELECT table_name, column_name, data_type, is_nullable, column_default "
+                "FROM information_schema.columns WHERE table_schema = %s AND table_name = ANY(%s)",
+                (schema, TE_TABLES))
+    sig["columns"] = sorted((t, c, d, n, norm(df)) for t, c, d, n, df in cur.fetchall())
+    cur.execute("SELECT r.relname, c.conname, c.contype, pg_get_constraintdef(c.oid) FROM pg_constraint c "
+                "JOIN pg_class r ON r.oid = c.conrelid WHERE r.relnamespace = %s::regnamespace "
+                "AND r.relname = ANY(%s)", (schema, TE_TABLES))
+    sig["constraints"] = sorted((t, n, k, norm(d)) for t, n, k, d in cur.fetchall())
+    cur.execute("SELECT tablename, indexname, indexdef FROM pg_indexes WHERE schemaname = %s "
+                "AND tablename = ANY(%s)", (schema, TE_TABLES))
+    sig["indexes"] = sorted((t, n, norm(d)) for t, n, d in cur.fetchall())
+    cur.execute("SELECT r.relname, t.tgname, pg_get_triggerdef(t.oid) FROM pg_trigger t "
+                "JOIN pg_class r ON r.oid = t.tgrelid WHERE r.relnamespace = %s::regnamespace "
+                "AND r.relname = ANY(%s) AND NOT t.tgisinternal", (schema, TE_TABLES))
+    sig["triggers"] = sorted((t, n, norm(d)) for t, n, d in cur.fetchall())
+    cur.execute("SELECT p.proname, pg_get_functiondef(p.oid) FROM pg_proc p WHERE "
+                "p.pronamespace = %s::regnamespace AND p.proname IN ('set_updated_at', 'te_fill_program_id')",
+                (schema,))
+    sig["functions"] = sorted((n, norm(d)) for n, d in cur.fetchall())
+    return sig
+
+
+@pytest.mark.integration
+class TenantTeSchemaUpgrades(unittest.TestCase):
+    def setUp(self):
+        self.tag = uuid.uuid4().hex[:10]
+        self.slug = f"tem_{self.tag}"
+        self.schema = schema_for(self.slug)
+        self.scratch: list[str] = []
+        self._template = settings.TE_SCHEMA
+        with owner_connect() as conn, conn.cursor() as cur:
+            self.template_sig = signature(cur, self._template)
+        self.assertTrue(self.template_sig["columns"], f"the template schema {self._template!r} is not deployed")
+
+    def tearDown(self):
+        settings.TE_SCHEMA = self._template
+        with owner_connect() as conn, conn.cursor() as cur:
+            for schema in [self.schema, *self.scratch]:
+                cur.execute(sql.SQL("DROP SCHEMA IF EXISTS {} CASCADE").format(sql.Identifier(schema)))
+            conn.commit()
+
+    # -- helpers ---------------------------------------------------------------------------
+
+    def _q(self, query, args=()):
+        """Run one statement and commit it; the rows if it returned any."""
+        with owner_connect() as conn, conn.cursor() as cur:
+            cur.execute(query, args)
+            rows = cur.fetchall() if cur.description else None
+            conn.commit()
+            return rows
+
+    def _signature(self, schema=None):
+        with owner_connect() as conn, conn.cursor() as cur:
+            return signature(cur, schema or self.schema)
+
+    def _version(self) -> str | None:
+        rows = self._q(sql.SQL("SELECT version_num FROM {}.{}").format(
+            sql.Identifier(self.schema), sql.Identifier(TE_VERSION_TABLE)))
+        return rows[0][0] if rows else None
+
+    def _assert_matches_template(self, schema=None):
+        actual = self._signature(schema)
+        for part, expected in self.template_sig.items():
+            missing, extra = sorted(set(expected) - set(actual[part])), sorted(set(actual[part]) - set(expected))
+            self.assertEqual((missing, extra), ([], []),
+                             f"{part}: the tenant lacks the first list and has the second beyond the template")
+
+    def _migrate(self) -> str:
+        with owner_connect() as conn, conn.cursor() as cur:
+            return migrate_te_schema(conn, cur, self.schema)
+
+    def _make_old_shape_tenant(self):
+        """A tenant as it was before #177, with the template's seed rows in it, and no version table."""
+        migrate_tenant(self.slug)
+        command.downgrade(_te_config(self.schema), TE_BASELINE)
+        with owner_connect() as conn, conn.cursor() as cur:
+            cur.execute(sql.SQL("DROP TABLE {}.{}").format(sql.Identifier(self.schema),
+                                                           sql.Identifier(TE_VERSION_TABLE)))
+            for table in TE_TABLES:
+                cur.execute("SELECT column_name FROM information_schema.columns WHERE table_schema = %s "
+                            "AND table_name = %s", (self.schema, table))
+                mine = {r[0] for r in cur.fetchall()}
+                cur.execute("SELECT column_name FROM information_schema.columns WHERE table_schema = %s "
+                            "AND table_name = %s", (self._template, table))
+                cols = sql.SQL(", ").join(sql.Identifier(c) for c in sorted(mine & {r[0] for r in cur.fetchall()}))
+                cur.execute(sql.SQL("INSERT INTO {s}.{t} ({c}) SELECT {c} FROM {tpl}.{t}").format(
+                    s=sql.Identifier(self.schema), t=sql.Identifier(table), c=cols,
+                    tpl=sql.Identifier(self._template)))
+            conn.commit()
+        seeded = self._q(sql.SQL("SELECT (SELECT count(*) FROM {t}.test_programs), (SELECT count(*) FROM {t}.vcrm_entries)")
+                         .format(t=sql.Identifier(self._template)))[0]
+        self.assertTrue(seeded[0] >= 2 and seeded[1] >= 1,
+                        f"the template needs seed rows for two programs and a VCRM entry (SEED_DEV=true); has {seeded}")
+
+    # -- paths to the template's shape -----------------------------------------------------
+
+    def test_a_new_tenant_is_cloned_stamped_at_head_and_matches_the_template(self):
+        migrate_tenant(self.slug)
+        self.assertEqual(self._version(), te_head())
+        self._assert_matches_template()
+
+    def test_the_template_is_stamped_with_the_head(self):
+        with owner_connect() as conn, conn.cursor() as cur:
+            self.assertEqual(template_te_version(cur, self._template), te_head())
+
+    def test_running_again_changes_nothing(self):
+        migrate_tenant(self.slug)
+        before = self._signature()
+        self.assertEqual(self._migrate(), "upgraded")
+        self.assertEqual(self._signature(), before)
+
+    def test_downgrade_restores_the_shape_from_before_177_and_upgrade_restores_the_template(self):
+        migrate_tenant(self.slug)
+        command.downgrade(_te_config(self.schema), TE_BASELINE)
+        sig = self._signature()
+        for table in _PROGRAM_TABLES:
+            self.assertNotIn((table, "program_id"), [(t, c) for t, c, *_ in sig["columns"]], table)
+        self.assertEqual([t for t in sig["triggers"] if t[1] == "trg_fill_program_id"], [])
+        self.assertEqual(sig["functions"][0][0], "set_updated_at")
+        self.assertEqual(len(sig["functions"]), 1, "te_fill_program_id must be gone")
+        fks = [d for t, n, k, d in sig["constraints"] if t == "test_cases" and k == "f" and "phase_id" in d]
+        self.assertEqual(fks, ["FOREIGN KEY (phase_id) REFERENCES <S>.test_phases(phase_id) ON DELETE CASCADE"])
+        command.upgrade(_te_config(self.schema), "head")
+        self._assert_matches_template()
+
+    def test_a_tenant_from_before_versioning_is_adopted_and_keeps_its_data(self):
+        self._make_old_shape_tenant()
+        updated_before = self._q(sql.SQL("SELECT tc_id, updated_at FROM {}.test_cases ORDER BY tc_id").format(
+            sql.Identifier(self.schema)))
+        self.assertTrue(updated_before)
+        self.assertEqual(self._migrate(), "adopted")
+        self.assertEqual(self._version(), te_head())
+        self._assert_matches_template()
+        # program_id came from each row's parent, so it equals the template's value for the same row
+        for table, key in (("test_cases", "tc_id"), ("test_events", "event_id"),
+                           ("vcrm_entries", "vcrm_id"), ("test_results", "result_id")):
+            mine = self._q(sql.SQL("SELECT {k}, program_id FROM {}.{} ORDER BY {k}").format(
+                sql.Identifier(self.schema), sql.Identifier(table), k=sql.Identifier(key)))
+            theirs = self._q(sql.SQL("SELECT {k}, program_id FROM {}.{} WHERE {k} = ANY(%s::uuid[]) ORDER BY {k}").format(
+                sql.Identifier(self._template), sql.Identifier(table), k=sql.Identifier(key)),
+                ([r[0] for r in mine],))
+            self.assertEqual(mine, theirs, table)
+        self.assertEqual(self._q(sql.SQL("SELECT tc_id, updated_at FROM {}.test_cases ORDER BY tc_id").format(
+            sql.Identifier(self.schema))), updated_before, "the backfill must not look like an edit")
+
+    def test_a_tenant_cloned_after_177_with_default_constraint_names_is_adopted_without_duplicates(self):
+        migrate_tenant(self.slug)
+        s = sql.Identifier(self.schema)
+        self._q(sql.SQL("ALTER TABLE {}.test_phases RENAME CONSTRAINT uq_phases_id_program "
+                        "TO test_phases_phase_id_program_id_key").format(s))
+        self._q(sql.SQL("DROP TABLE {}.{}").format(s, sql.Identifier(TE_VERSION_TABLE)))
+        self.assertEqual(self._migrate(), "adopted")
+        self._assert_matches_template()
+        unique_on_phases = self._q(sql.SQL(
+            "SELECT count(*) FROM pg_constraint WHERE conrelid = %s::regclass AND contype = 'u' "
+            "AND pg_get_constraintdef(oid) = 'UNIQUE (phase_id, program_id)'"), (f'"{self.schema}".test_phases',))
+        self.assertEqual(unique_on_phases[0][0], 1, "an equivalent constraint must be renamed, not duplicated")
+
+    def test_a_clone_carries_the_templates_index_and_constraint_names(self):
+        migrate_tenant(self.slug)
+        names = {n for _, n, *_ in self._signature()["indexes"]}
+        self.assertLessEqual({"idx_testcases_phase", "uq_phases_id_program", "idx_personnel_email_trgm"}, names)
+        # what LIKE ... INCLUDING ALL would have called the same objects
+        self.assertTrue(names.isdisjoint({"test_cases_phase_id_idx", "test_phases_phase_id_program_id_key",
+                                          "personnel_email_idx"}))
+
+    def test_cross_program_rows_stop_the_upgrade_name_the_rows_and_change_nothing(self):
+        self._make_old_shape_tenant()
+        s = sql.Identifier(self.schema)
+        # old shape: a VCRM entry may name a requirement from one program and a test case from another
+        (vcrm_id, other_req), = self._q(sql.SQL(
+            "SELECT v.vcrm_id, r.req_id FROM {s}.vcrm_entries v "
+            "JOIN {s}.test_cases c ON c.tc_id = v.tc_id JOIN {s}.test_phases ph ON ph.phase_id = c.phase_id "
+            "JOIN {s}.requirements r ON r.program_id <> ph.program_id LIMIT 1").format(s=s))
+        self._q(sql.SQL("UPDATE {}.vcrm_entries SET req_id = %s WHERE vcrm_id = %s").format(s), (other_req, vcrm_id))
+        before = self._signature()
+        with self.assertRaises(RuntimeError) as ctx:
+            self._migrate()
+        self.assertIn("Nothing was changed", str(ctx.exception))
+        self.assertIn(str(vcrm_id), str(ctx.exception))
+        self.assertIn(self.schema, str(ctx.exception))
+        self.assertEqual(self._signature(), before, "the failed upgrade must roll back completely")
+        self.assertEqual(self._version(), TE_BASELINE)
+
+    def test_a_template_at_an_older_version_is_cloned_then_brought_forward(self):
+        scratch = f"te_tpl_{self.tag}"
+        self.scratch.append(scratch)
+        with owner_connect() as conn, conn.cursor() as cur:
+            cur.execute(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(scratch)))
+            provision_te_tables(cur, scratch, self._template)
+            conn.commit()
+        command.stamp(_te_config(scratch), te_head())
+        command.downgrade(_te_config(scratch), TE_BASELINE)
+        self._q(sql.SQL("COMMENT ON SCHEMA {} IS %s").format(sql.Identifier(scratch)),
+                (f"te_schema_version={TE_BASELINE}",))
+        settings.TE_SCHEMA = scratch
+        try:
+            self.assertEqual(self._migrate_new(), "provisioned")
+        finally:
+            settings.TE_SCHEMA = self._template
+        self.assertEqual(self._version(), te_head())
+        self._assert_matches_template()
+
+    def _migrate_new(self) -> str:
+        """migrate_te_schema for a tenant that has no T&E tables yet (its schema is made if need be)."""
+        self._q(sql.SQL("CREATE SCHEMA IF NOT EXISTS {}").format(sql.Identifier(self.schema)))
+        return self._migrate()
+
+    def test_the_next_schema_change_reaches_existing_and_new_tenants(self):
+        """The reproduction in #184: change the T&E tables after a tenant exists and it must get the change.
+
+        A throwaway migration, te9001, is added on top of the real chain (in a temporary copy of it).
+        The template is still at te0002, so a new tenant is cloned from it and brought forward too.
+        """
+        existing = self.slug
+        migrate_tenant(existing)
+        chain = Path(tempfile.mkdtemp(prefix="alembic_te_next_"))
+        try:
+            shutil.copytree(ROOT / "alembic_te", chain, dirs_exist_ok=True,
+                            ignore=shutil.ignore_patterns("__pycache__"))
+            (chain / "versions" / "te9001_demo.py").write_text(
+                'from alembic import op\nrevision = "te9001"\ndown_revision = "te0002"\n'
+                "branch_labels = None\ndepends_on = None\n\n\ndef upgrade():\n"
+                '    schema = op.get_context().opts["version_table_schema"]\n'
+                '    op.execute(f\'ALTER TABLE "{schema}".organisations ADD COLUMN demo_note TEXT\')\n\n\n'
+                "def downgrade():\n    pass\n", encoding="utf-8")
+
+            def config_with_next_change(schema):
+                cfg = _te_config(schema)
+                cfg.set_main_option("script_location", str(chain))
+                return cfg
+
+            def has_demo_note(schema):
+                return bool(self._q("SELECT 1 FROM information_schema.columns WHERE table_schema = %s "
+                                    "AND table_name = 'organisations' AND column_name = 'demo_note'", (schema,)))
+
+            with mock.patch("api.tenancy._te_config", side_effect=config_with_next_change):
+                self.assertFalse(has_demo_note(self.schema))
+                self.assertEqual(self._migrate(), "upgraded")
+                self.assertTrue(has_demo_note(self.schema), "an existing tenant did not get the change")
+                self.assertEqual(self._version(), "te9001")
+
+                newcomer = f"tenant_tem_{uuid.uuid4().hex[:10]}"
+                self.scratch.append(newcomer)
+                self._q(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(newcomer)))
+                with owner_connect() as conn, conn.cursor() as cur:
+                    self.assertEqual(migrate_te_schema(conn, cur, newcomer), "provisioned")
+                self.assertTrue(has_demo_note(newcomer), "a tenant cloned from the older template did not get it")
+        finally:
+            shutil.rmtree(chain, ignore_errors=True)
+
+    def test_a_template_with_no_stamp_is_refused_with_the_way_out(self):
+        scratch = f"te_tpl_{self.tag}"
+        self.scratch.append(scratch)
+        with owner_connect() as conn, conn.cursor() as cur:
+            cur.execute(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(scratch)))
+            provision_te_tables(cur, scratch, self._template)
+            conn.commit()
+        settings.TE_SCHEMA = scratch
+        with self.assertRaisesRegex(RuntimeError, "Redeploy it with build/deploy_all.sh"):
+            self._migrate_new()
+        self.assertEqual(self._q("SELECT to_regclass(%s)", (f'"{self.schema}".test_cases',))[0][0], None)
+
+    def test_a_template_stamped_with_an_unknown_version_is_refused(self):
+        scratch = f"te_tpl_{self.tag}"
+        self.scratch.append(scratch)
+        with owner_connect() as conn, conn.cursor() as cur:
+            cur.execute(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(scratch)))
+            provision_te_tables(cur, scratch, self._template)
+            cur.execute(sql.SQL("COMMENT ON SCHEMA {} IS 'te_schema_version=te9999'").format(sql.Identifier(scratch)))
+            conn.commit()
+        settings.TE_SCHEMA = scratch
+        with self.assertRaisesRegex(RuntimeError, self.schema):
+            self._migrate_new()
+
+    def test_two_workers_starting_together_leave_one_correct_tenant(self):
+        self._q(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(self.schema)))  # the T&E tables are what both race to build
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            results = [f.result() for f in [pool.submit(self._migrate_new) for _ in range(2)]]
+        self.assertEqual(sorted(results), ["provisioned", "upgraded"])
+        self.assertEqual(self._version(), te_head())
+        self._assert_matches_template()
diff --git a/tests/system/test_type_provenance.py b/tests/system/test_type_provenance.py
new file mode 100644
index 0000000..554e9b1
--- /dev/null
+++ b/tests/system/test_type_provenance.py
@@ -0,0 +1,112 @@
+"""Dynamic-mode tables are created from confirmed types, and the registry says whether they were (#195).
+
+Decided 4 Nov: preview creates nothing; the table appears only when the upload is made, with the
+types sent; the registry records ``user_confirmed`` or ``inferred_unconfirmed``; and a dynamic
+table can never be the source of a T&E import.
+
+Integration: needs a live PostgreSQL. Fails (never skips) without one.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import unittest
+import uuid
+
+import pytest
+from fastapi.testclient import TestClient
+
+from api.config import settings
+from api.db import Conn
+from api.main import app
+
+_HELP = ("Start PostgreSQL and ensure api/ can reach it (PGHOST/PGPORT/PGUSER/"
+         "PGPASSWORD/PGDATABASE). See scripts/start-api.ps1.")
+
+
+def _column_types(table: str) -> dict[str, str]:
+    with Conn() as conn, conn.cursor() as cur:
+        cur.execute("SELECT column_name, data_type FROM information_schema.columns "
+                    "WHERE table_schema = %s AND table_name = %s", (settings.UPLOADS_SCHEMA, table))
+        return dict(cur.fetchall())
+
+
+@pytest.mark.integration
+class TypeConfirmationAndProvenance(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.ctx = TestClient(app)
+        try:
+            cls.client = cls.ctx.__enter__()
+        except Exception as exc:  # noqa: BLE001
+            raise AssertionError(f"Could not start the API: {type(exc).__name__}: {exc}\n{_HELP}") from None
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.ctx.__exit__(None, None, None)
+
+    def setUp(self):
+        self.tag = f"{os.getpid()}_{uuid.uuid4().hex[:8]}"
+        self.name = f"prov_{self.tag}.csv"
+        self.content = f"n,label\n1,{self.tag}\n"
+
+    def _upload(self, **extra) -> dict:
+        body = {"fileName": self.name, "content": self.content, **extra}
+        return self.client.post("/api/csv/upload", json=body).json()
+
+    def _registered(self) -> dict | None:
+        return next((f for f in self.client.get("/api/csv/files").json() if f["file_name"] == self.name), None)
+
+    def test_preview_proposes_types_but_creates_no_table_and_no_registry_row(self):
+        preview = self.client.post("/api/csv/preview", json={"fileName": self.name, "content": self.content}).json()
+        self.assertEqual(preview["status"], "ok", preview)
+        self.assertEqual(len(preview["inferredTypes"]), 2)
+        table = f"csv_{hashlib.sha256(self.content.encode()).hexdigest()[:16]}"
+        self.assertEqual(_column_types(table), {}, "preview must not create the table")
+        self.assertIsNone(self._registered())
+
+    def test_confirming_creates_the_table_with_the_confirmed_types(self):
+        res = self._upload(types=["int8", "text"], typesConfirmed=True)
+        self.assertEqual(res["status"], "ok", res)
+        self.assertEqual(res["typesProvenance"], "user_confirmed")
+        self.assertEqual(_column_types(res["tableName"]).get("n"), "bigint")
+        self.assertEqual(self._registered()["types_provenance"], "user_confirmed")
+
+    def test_the_confirmed_types_can_differ_from_what_was_inferred(self):
+        res = self._upload(types=["text", "text"], typesConfirmed=True)
+        self.assertEqual(_column_types(res["tableName"]).get("n"), "text")
+
+    def test_types_sent_without_confirmation_are_applied_but_not_marked_confirmed(self):
+        res = self._upload(types=["int8", "text"])
+        self.assertEqual(res["typesProvenance"], "inferred_unconfirmed")
+        self.assertEqual(_column_types(res["tableName"]).get("n"), "bigint")
+        self.assertEqual(self._registered()["types_provenance"], "inferred_unconfirmed")
+
+    def test_no_types_at_all_is_unconfirmed(self):
+        res = self._upload()
+        self.assertEqual((res["status"], res["typesProvenance"]), ("ok", "inferred_unconfirmed"))
+        self.assertEqual(self._registered()["types_provenance"], "inferred_unconfirmed")
+
+    def test_confirmation_without_the_confirmed_types_is_rejected_and_loads_nothing(self):
+        resp = self.client.post("/api/csv/upload", json={
+            "fileName": self.name, "content": self.content, "typesConfirmed": True})
+        self.assertEqual(resp.status_code, 422)
+        self.assertIsNone(self._registered())
+
+    def test_te_mode_rejects_types_confirmed(self):
+        resp = self.client.post("/api/csv/upload", json={
+            "fileName": self.name, "content": self.content, "mode": "te",
+            "targetTable": "organisations", "typesConfirmed": True})
+        self.assertEqual(resp.status_code, 422)
+        self.assertIn("dynamic", resp.json()["detail"])
+
+    def test_te_import_rejects_a_dynamic_table_as_its_target(self):
+        dynamic = self._upload(types=["int8", "text"], typesConfirmed=True)
+        self.assertEqual(dynamic["status"], "ok", dynamic)
+        res = self.client.post("/api/csv/upload", json={
+            "fileName": f"te_{self.name}", "content": self.content, "mode": "te",
+            "targetTable": dynamic["tableName"]}).json()
+        self.assertEqual(res["status"], "error", res)
+        self.assertIn(dynamic["tableName"], res["message"].lower())
+        self.assertFalse([f for f in self.client.get("/api/csv/files").json() if f["file_name"] == f"te_{self.name}"])
diff --git a/tests/unit/test_prod_db_naming.py b/tests/unit/test_prod_db_naming.py
new file mode 100644
index 0000000..1804c1d
--- /dev/null
+++ b/tests/unit/test_prod_db_naming.py
@@ -0,0 +1,234 @@
+"""Terraform, the deployer and the container entrypoint agree on the production database (#172).
+
+Before: Terraform created `te_prod` owned by `pgadmin`, the deployer created `te_mgmt_prod` owned
+by `postgres`, and the entrypoint expected an `env_prod.sql` that nothing in the image produced.
+
+No database needed: the deployer runs with --dry-run, the entrypoint with its `render` action, and
+the Terraform file is read as text. The fresh-server run is recorded on the PR.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+BUILD = ROOT / "build"
+ENTRYPOINT = ROOT / "infra" / "entrypoint.sh"
+TERRAFORM = (ROOT / "infra" / "terraform-prod" / "main.tf").read_text(encoding="utf-8")
+SCHEMA_SQL = BUILD / "te_core_schema.sql"
+SECRET = "S3cr3t'\"$(touch /tmp/should-not-exist)"
+
+
+def _find_bash():
+    """A real bash. On Windows the PATH `bash` is the WSL launcher, which fails when no distribution is
+    installed, so use Git Bash explicitly (as test_csv_utilise.py does). None if there isn't one."""
+    if sys.platform == "win32":
+        for candidate in (r"C:\Program Files\Git\bin\bash.exe", r"C:\Program Files (x86)\Git\bin\bash.exe"):
+            if Path(candidate).exists():
+                return candidate
+        found = shutil.which("bash")
+        return found if found and "system32" not in found.lower() else None
+    return shutil.which("bash") or "bash"
+
+
+_BASH = _find_bash()
+
+
+def _posix(path: Path) -> str:
+    """Forward slashes, which Git Bash and Linux bash both read; Windows backslashes are escapes to bash."""
+    return path.as_posix()
+
+
+def _bash(args: list[str], env: dict[str, str | None]) -> subprocess.CompletedProcess:
+    """Run bash with the caller's environment plus ``env``; a None value removes that variable."""
+    full_env = dict(os.environ)  # Git Bash on Windows needs SYSTEMROOT, TEMP and the like
+    for key, value in env.items():
+        if value is None:
+            full_env.pop(key, None)
+        else:
+            full_env[key] = value
+    # The scripts print UTF-8 symbols; the Windows locale codec cannot decode them.
+    return subprocess.run([_BASH, *args], env=full_env, capture_output=True, encoding="utf-8",
+                          errors="replace", timeout=60)
+
+
+def _require_bash() -> None:
+    if _BASH is None:
+        raise AssertionError("No working bash found: install Git Bash and ensure it is on PATH.")
+
+
+def _tf_env(name: str) -> str | None:
+    m = re.search(r'env\s*\{\s*name\s*=\s*"%s"\s*(?:value|secret_name)\s*=\s*([^\n]+)' % name, TERRAFORM)
+    return m.group(1).strip().strip('"') if m else None
+
+
+@pytest.mark.unit
+class DeployerReadsNamesFromTheLauncher(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        _require_bash()
+
+    def setUp(self):
+        self.tmp = Path(tempfile.mkdtemp(prefix="deploy_names_"))
+        (self.tmp / "build").mkdir()
+        shutil.copy(BUILD / "deploy_all.sh", self.tmp / "build" / "deploy_all.sh")
+        shutil.copytree(BUILD / "environments", self.tmp / "build" / "environments")
+        for env in ("dev", "test", "staging", "prod"):
+            shutil.copy(self.tmp / "build" / "environments" / f"env_{env}.example.sql",
+                        self.tmp / "build" / "environments" / f"env_{env}.sql")
+
+    def tearDown(self):
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def _launcher(self, env: str) -> Path:
+        return self.tmp / "build" / "environments" / f"env_{env}.sql"
+
+    def _dry_run(self, *envs: str) -> subprocess.CompletedProcess:
+        # PGHOST is unreachable on purpose: a dry run must not touch PostgreSQL.
+        return _bash([_posix(self.tmp / "build" / "deploy_all.sh"), "--dry-run", *envs],
+                     {"PGHOST": "/nonexistent-socket-dir"})
+
+    def _write_launcher(self, env: str, text: str) -> None:
+        self._launcher(env).write_text(text, encoding="utf-8", newline="\n")
+
+    def test_committed_templates_keep_the_on_prem_names(self):
+        out = self._dry_run().stdout
+        for env, db in (("DEV", "te_mgmt_dev"), ("TEST", "te_mgmt_test"),
+                        ("STAGING", "te_mgmt_staging"), ("PROD", "te_mgmt_prod")):
+            self.assertRegex(out, rf"{env}\s+\(database: {db}, owner: postgres\)")
+
+    def test_the_launcher_decides_the_database_and_owner(self):
+        text = self._launcher("prod").read_text(encoding="utf-8")
+        text = re.sub(r"(\\set db_name\s+).*", r"\1te_prod", text)
+        text = re.sub(r"(\\set db_owner\s+).*", r"\1pgadmin", text)
+        self._write_launcher("prod", text)
+        res = self._dry_run("prod")
+        self.assertEqual(res.returncode, 0, res.stderr)
+        self.assertRegex(res.stdout, r"PROD\s+\(database: te_prod, owner: pgadmin\)")
+
+    def test_a_launcher_that_sets_neither_falls_back_to_the_on_prem_names(self):
+        lines = [ln for ln in self._launcher("prod").read_text(encoding="utf-8").splitlines()
+                 if not re.match(r"\\set (db_name|db_owner)\b", ln)]
+        self._write_launcher("prod", "\n".join(lines))
+        self.assertRegex(self._dry_run("prod").stdout, r"database: te_mgmt_prod, owner: postgres")
+
+    def test_an_unsafe_identifier_is_refused(self):
+        text = re.sub(r"(\\set db_name\s+).*", r'\1"x; DROP DATABASE y"',
+                      self._launcher("prod").read_text(encoding="utf-8"))
+        self._write_launcher("prod", text)
+        res = self._dry_run("prod")
+        self.assertNotEqual(res.returncode, 0)
+        self.assertIn("Unsafe identifier", res.stderr)
+
+
+@pytest.mark.unit
+class EntrypointWritesTheLauncher(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        _require_bash()
+
+    def setUp(self):
+        self.home = Path(tempfile.mkdtemp(prefix="entrypoint_home_"))
+        # The image carries the templates only; the concrete launchers are gitignored and absent.
+        (self.home / "build" / "environments").mkdir(parents=True)
+        for template in (BUILD / "environments").glob("env_*.example.sql"):
+            shutil.copy(template, self.home / "build" / "environments" / template.name)
+        self.env = {"MIGRATION_HOME": _posix(self.home), "TARGET_ENV": "prod", "PGHOST": "unused",
+                    "PGPORT": "5432", "PGUSER": "pgadmin", "PGDATABASE": "te_prod",
+                    "WAIT_FOR_DB_SECONDS": "1", "APP_DB_PASSWORD": SECRET}
+
+    def tearDown(self):
+        shutil.rmtree(self.home, ignore_errors=True)
+        self.assertFalse(Path("/tmp/should-not-exist").exists(), "the password was executed as shell")
+
+    def _render(self, **override) -> subprocess.CompletedProcess:
+        return _bash([_posix(ENTRYPOINT), "render"], {**self.env, **override})
+
+    def test_the_image_has_no_concrete_launcher(self):
+        self.assertEqual([p.name for p in (self.home / "build" / "environments").glob("env_*.sql")
+                          if not p.name.endswith(".example.sql")], [])
+
+    def test_the_launcher_is_made_from_the_containers_database_and_user(self):
+        res = self._render()
+        self.assertEqual(res.returncode, 0, res.stderr)
+        self.assertRegex(res.stdout, r"\\set db_name\s+te_prod\b")
+        self.assertRegex(res.stdout, r"\\set db_owner\s+pgadmin\b")
+        self.assertIn(r"\ir ../te_core_schema.sql", res.stdout)
+
+    def test_the_password_is_read_by_psql_and_never_written(self):
+        res = self._render()
+        self.assertIn(r"\getenv app_password APP_DB_PASSWORD", res.stdout)
+        written = (self.home / "build" / "environments" / "env_prod.sql").read_text(encoding="utf-8")
+        self.assertNotIn("S3cr3t", written)
+        self.assertNotIn("S3cr3t", res.stdout + res.stderr)
+        self.assertNotIn("__INJECT_AT_DEPLOY__", written)
+
+    def test_a_missing_password_is_an_error(self):
+        res = self._render(APP_DB_PASSWORD=None)
+        self.assertNotEqual(res.returncode, 0)
+        self.assertIn("APP_DB_PASSWORD is not set", res.stderr)
+
+    def test_an_unsafe_database_name_is_an_error(self):
+        res = self._render(PGDATABASE="te_prod; DROP DATABASE x")
+        self.assertNotEqual(res.returncode, 0)
+        self.assertIn("lower-case letters", res.stderr)
+
+    def test_an_existing_launcher_is_left_alone(self):
+        mine = self.home / "build" / "environments" / "env_prod.sql"
+        mine.write_text("\\set db_name mounted_db\n", encoding="utf-8")
+        res = self._render()
+        self.assertEqual(res.returncode, 0, res.stderr)
+        self.assertEqual(mine.read_text(encoding="utf-8"), "\\set db_name mounted_db\n")
+
+    def test_an_environment_with_no_template_is_an_error(self):
+        res = self._render(TARGET_ENV="qa")
+        self.assertNotEqual(res.returncode, 0)
+        self.assertIn("no launcher for TARGET_ENV='qa'", res.stderr)
+
+
+@pytest.mark.unit
+class TerraformAgreesWithTheDeployment(unittest.TestCase):
+    def test_the_database_terraform_creates_is_te_prod(self):
+        m = re.search(r'resource "azurerm_postgresql_flexible_server_database" "te_prod"\s*\{[^}]*?name\s*=\s*"([^"]+)"',
+                      TERRAFORM)
+        self.assertIsNotNone(m)
+        self.assertEqual(m.group(1), "te_prod")
+
+    def test_the_job_deploys_into_that_database_as_the_servers_admin(self):
+        admin = re.search(r'administrator_login\s*=\s*"([^"]+)"', TERRAFORM).group(1)
+        self.assertEqual(_tf_env("PGUSER"), admin)
+        self.assertEqual(_tf_env("PGDATABASE"), "azurerm_postgresql_flexible_server_database.te_prod.name")
+
+    def test_the_job_receives_the_application_password_as_a_secret(self):
+        self.assertEqual(_tf_env("APP_DB_PASSWORD"), "app-password")
+        self.assertRegex(TERRAFORM, r'secret\s*\{\s*name\s*=\s*"app-password"[^}]*azurerm_key_vault_secret\.app_password\.id')
+
+    def test_every_extension_the_schema_creates_is_allow_listed_on_azure(self):
+        needed = set(re.findall(r'CREATE EXTENSION IF NOT EXISTS "([^"]+)"', SCHEMA_SQL.read_text(encoding="utf-8")))
+        self.assertTrue(needed, "expected the schema to create extensions")
+        m = re.search(r'name\s*=\s*"azure\.extensions"[^}]*?value\s*=\s*"([^"]+)"', TERRAFORM)
+        self.assertIsNotNone(m, "azure.extensions is not set")
+        self.assertLessEqual(needed, set(m.group(1).split(",")))
+
+
+@pytest.mark.unit
+class DeploymentOutputDoesNotEchoThePassword(unittest.TestCase):
+    def test_the_two_schema_copies_are_identical(self):
+        self.assertEqual(SCHEMA_SQL.read_bytes(), (BUILD / "schema" / "postgresql" / "te_core_schema.sql").read_bytes())
+
+    def test_the_password_row_is_captured_not_printed(self):
+        text = SCHEMA_SQL.read_text(encoding="utf-8")
+        start = text.index("set_config('te.env_label'")
+        stmt = text[start:text.index("\\echo", start)]
+        self.assertIn("te.app_password", stmt)
+        self.assertIn("\\gset", stmt)
+        self.assertNotIn(";", stmt, "a ';'-terminated SELECT prints the row, password included")