diff --git a/docs/changelog.rst b/docs/changelog.rst
index 30ad9899e..9228fb6e1 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -18,6 +18,20 @@ Unreleased
 - Fixed an ``IndexError`` from ``table.insert(..., pk=..., ignore=True)`` when an ignored insert followed writes to another table on the same connection. ``last_pk`` is now populated from the explicit primary key value instead of looking up a stale ``lastrowid``. (:issue:`554`)
 - Fixed a bug where a failed write statement executed with ``db.execute()`` left the driver's implicit transaction open. Every subsequent write then joined that phantom transaction, which nothing committed, so their work was silently rolled back when the connection was closed. The implicit transaction opened by a failed statement is now rolled back before the exception is raised. A failed write inside a transaction opened with ``db.begin()`` or ``db.atomic()`` leaves that transaction open and untouched, as before.
 - Fixed a bug where transaction-control statements prefixed with an empty statement - ``db.query("; COMMIT")`` - or a UTF-8 byte order mark slipped past the check that rejects them, committing the caller's open transaction before raising a confusing ``OperationalError``. The keyword scanner used by ``db.query()`` and ``db.execute()`` now skips leading ``;`` and byte order marks, matching what the ``sqlite3`` driver tolerates before the first token, so these statements are rejected with a ``ValueError`` without being executed. The same fix means ``db.execute("; BEGIN")`` no longer auto-commits the transaction it just opened.
+- Documented a limitation of ``db.query()``: a ``PRAGMA`` statement that returns no rows raises a ``ValueError`` but still takes effect, because PRAGMA statements run outside the savepoint guard used to roll back other rejected statements. Use ``db.execute()`` for row-less PRAGMA statements.
+- Fixed exception masking when a statement destroys the enclosing transaction. An error such as a ``RAISE(ROLLBACK)`` trigger or ``INSERT OR ROLLBACK`` conflict rolls back the whole transaction, destroying every savepoint - the cleanup in ``db.atomic()`` and ``db.query()`` then failed with ``OperationalError: no such savepoint`` (or ``cannot rollback - no transaction is active``), hiding the original ``IntegrityError`` from code that tried to catch it. Cleanup now checks whether a transaction is still open first, so the original exception propagates.
+- ``sqlite-utils migrate --list`` is now read-only even when the migrations file uses the legacy ``sqlite_migrate.Migrations`` class, whose listing methods create the ``_sqlite_migrations`` table as a side effect. The listing now runs inside a transaction that is rolled back.
+- ``sqlite-utils insert ... --pk <missing column>`` and ``sqlite-utils extract <missing column>`` now show a clean ``Error:`` message instead of a raw Python traceback. The ``extract`` command also shows a clean error when pointed at a view.
+- Fixed a bug where running ``table.extract()`` more than once against the same lookup table inserted duplicate rows for values containing ``null`` - SQLite unique indexes treat ``NULL`` values as distinct, so ``INSERT OR IGNORE`` alone could not dedupe them. Each repeat extract added another copy that nothing referenced. The insert now uses an ``IS``-based ``NOT EXISTS`` guard so ``null``-containing rows match existing lookup rows.
+- ``db.add_foreign_keys()`` no longer silently ignores requested ``ON DELETE``/``ON UPDATE`` actions when a foreign key with the same columns already exists - it raises ``AlterError`` suggesting ``table.transform()``, since the actions of an existing foreign key cannot be changed in place. Exact duplicates, including actions, are still skipped so repeated calls stay idempotent. The method also now validates that compound foreign keys have the same number of columns on both sides, instead of silently discarding the extra columns.
+- ``db.ensure_autocommit_on()`` now raises ``TransactionError`` if called while a transaction is open. Assigning ``isolation_level`` commits any pending transaction as a side effect, so entering the block silently committed the caller's open transaction and made a later ``rollback()`` a no-op.
+- ``sqlite-utils migrate --stop-before`` now exits with an error if the named migration has already been applied. Previously the name passed validation but was only checked against pending migrations, so every migration after it was silently applied - the exact outcome ``--stop-before`` exists to prevent. ``Migrations.apply(db, stop_before=...)`` raises ``ValueError`` in the same situation, before applying anything.
+- Fixed a regression where ``table.insert(..., pk=..., alter=True)`` raised ``InvalidColumns`` if the primary key column did not exist in the table yet. With ``alter=True`` the check now waits until the record keys are known, so a pk column supplied by the records is added by the alter as it was in 3.x. A pk column found in neither the table nor the records still raises ``InvalidColumns``.
+- Fixed a bug where inserting CSV or TSV data into an existing table rewrote that table's column types to match the incoming file. Type detection is the default in 4.0, so ``sqlite-utils insert data.db places places.csv --csv`` against a table with a ``TEXT`` zip code column would convert the column to ``INTEGER`` and corrupt values with leading zeros - ``"01234"`` became ``1234``. Detected types are now only applied when the ``insert`` or ``upsert`` command creates the table.
+- Fixed ``pks_and_rows_where()`` raising ``AttributeError`` when called on a view, and no longer double-quotes the synthesized ``rowid`` column in its generated SQL - SQLite turns a double-quoted identifier that does not resolve into a string literal, which on a view produced a confusing ``KeyError`` instead of the ``OperationalError`` raised in 3.x. Compound primary keys returned by this method now follow ``PRIMARY KEY`` declaration order.
+- The ``foreign_keys=`` argument to ``create()`` and ``insert()`` accepts a mixed list of ``ForeignKey`` objects, tuples and column name strings again. In 4.0 pre-releases mixing ``ForeignKey`` objects with tuples raised a ``ValueError`` - a regression from 3.x, where ``ForeignKey`` was a ``namedtuple`` and passed the tuple checks.
+- ``ForeignKey`` objects are hashable again. The 4.0 change from ``namedtuple`` to dataclass accidentally made them unhashable, breaking patterns like ``set(table.foreign_keys)`` that worked in 3.x. ``ForeignKey`` is now a frozen dataclass - immutable and hashable, like the namedtuple was.
+- Fixed a bug where compound primary key columns were returned in table column order instead of ``PRIMARY KEY`` declaration order. For a table declared as ``CREATE TABLE other (b TEXT, a TEXT, PRIMARY KEY (a, b))`` an implicit ``FOREIGN KEY (x, y) REFERENCES other`` was introspected as referencing ``(b, a)`` when SQLite resolves it as ``(a, b)`` - running ``transform()`` on such a table then rewrote the schema with the inverted column order, silently reversing the meaning of the constraint and causing foreign key errors on valid data. ``table.pks``, compound foreign key guessing and ``transform()`` now all use the primary key declaration order, and ``transform()`` no longer reorders a compound ``PRIMARY KEY (b, a)`` into table column order.
 
 .. _v4_0rc3:
 
diff --git a/docs/cli.rst b/docs/cli.rst
index 1f95bbde0..063e80f48 100644
--- a/docs/cli.rst
+++ b/docs/cli.rst
@@ -1326,6 +1326,8 @@ A progress bar is displayed when inserting data from a file. You can hide the pr
 
 By default, column types are automatically detected for CSV or TSV files - resulting in a mix of ``TEXT``, ``INTEGER`` and ``REAL`` columns. To disable type detection and treat all columns as ``TEXT``, use the ``--no-detect-types`` option.
 
+Detected types are only applied when the table is created by the command. Inserting CSV or TSV data into a table that already exists leaves the existing column types unchanged - values are inserted using the table's existing schema.
+
 For example, given a ``creatures.csv`` file containing this:
 
 .. code-block::
diff --git a/docs/migrations.rst b/docs/migrations.rst
index e685936e4..cfdbf138a 100644
--- a/docs/migrations.rst
+++ b/docs/migrations.rst
@@ -157,7 +157,7 @@ You can also target a specific migration set using ``migration_set:migration_nam
 
 The ``--stop-before`` option can be passed more than once.
 
-If a ``--stop-before`` value does not match any known migration the command exits with an error, rather than silently applying everything.
+If a ``--stop-before`` value does not match any known migration the command exits with an error, rather than silently applying everything. Naming a migration that has already been applied is also an error - stopping before it is impossible to honor - and no pending migrations are applied.
 
 Verbose output
 ==============
diff --git a/docs/python-api.rst b/docs/python-api.rst
index 0e61cd46d..11af1f4e8 100644
--- a/docs/python-api.rst
+++ b/docs/python-api.rst
@@ -233,6 +233,8 @@ The SQL query is executed as soon as ``db.query()`` is called. The resulting row
 
 ``db.query()`` can only be used with SQL that returns rows. Passing a statement that returns no rows - an ``INSERT`` or ``UPDATE`` without a ``RETURNING`` clause, for example - will raise a ``ValueError``. The rejected statement is rolled back, so it has no effect on the database. Use :ref:`db.execute() <python_api_execute>` for those statements instead.
 
+There is one exception to the rolled-back guarantee: a ``PRAGMA`` statement that returns no rows, such as ``PRAGMA user_version = 5``, still raises a ``ValueError`` but will already have taken effect. Some PRAGMA statements refuse to run inside a transaction, so PRAGMAs are executed outside the savepoint that is used to roll back other rejected statements. Use ``db.execute()`` for PRAGMA statements that do not return rows.
+
 If a query returns more than one column with the same name - a join between two tables that share column names, for example - later occurrences are renamed with a numeric suffix, so every value is included in the dictionary:
 
 .. code-block:: python
@@ -1634,6 +1636,8 @@ Here's an example adding two foreign keys at once:
 
 This method runs the same checks as ``.add_foreign_keys()`` and will raise ``sqlite_utils.db.AlterError`` if those checks fail.
 
+Foreign keys that already exist are silently skipped, so repeated calls are idempotent - but only if they match exactly. Requesting a foreign key that exists with different ``ON DELETE``/``ON UPDATE`` actions raises ``AlterError``: use ``table.transform()`` to change the actions of an existing foreign key.
+
 .. _python_api_index_foreign_keys:
 
 Adding indexes for all foreign keys
diff --git a/docs/upgrading.rst b/docs/upgrading.rst
index 88e987d7e..92b582a45 100644
--- a/docs/upgrading.rst
+++ b/docs/upgrading.rst
@@ -95,7 +95,7 @@ Python API changes
     for fk in db["courses"].foreign_keys:
         fk.table, fk.column, fk.other_table, fk.other_column
 
-Attempting the old unpacking or ``fk[0]`` indexing now raises ``TypeError``, so any code using those patterns will fail loudly rather than silently misbehave.
+Attempting the old unpacking or ``fk[0]`` indexing now raises ``TypeError``, so any code using those patterns will fail loudly rather than silently misbehave. Like the old namedtuple, ``ForeignKey`` instances are immutable and hashable - they can be collected into sets and used as dictionary keys. Note that equality now includes the ``on_delete`` and ``on_update`` actions: a ``ForeignKey`` with ``ON DELETE CASCADE`` is not equal to one without.
 
 Compound foreign keys - previously returned as one ``ForeignKey`` per column, misleadingly suggesting several independent single-column keys - are now returned as a single ``ForeignKey`` with ``is_compound=True``. For these the scalar ``column`` and ``other_column`` fields are ``None``; use the ``columns`` and ``other_columns`` tuples instead. Single-column foreign keys are unaffected apart from the class change: ``column``/``other_column`` behave as before and ``columns``/``other_columns`` are one-item tuples.
 
diff --git a/sqlite_utils/cli.py b/sqlite_utils/cli.py
index 8ee20913f..fe7dbe4bd 100644
--- a/sqlite_utils/cli.py
+++ b/sqlite_utils/cli.py
@@ -13,6 +13,7 @@
     BadMultiValues,
     DEFAULT,
     DescIndex,
+    InvalidColumns,
     NoTable,
     NoView,
     quote_identifier,
@@ -1165,11 +1166,14 @@ def insert_upsert_implementation(
                     db.conn.cursor().executemany(bulk_sql, doc_chunk)
             return
 
+        # table_names() rather than db.table(), which raises NoTable for
+        # views before the error handling below can deal with them
+        table_existed_before_insert = table in db.table_names()
         try:
             db.table(table).insert_all(
                 docs, pk=pk, batch_size=batch_size, alter=alter, **extra_kwargs
             )
-        except NoTable as e:
+        except (NoTable, InvalidColumns) as e:
             raise click.ClickException(str(e))
         except Exception as e:
             if (
@@ -1194,7 +1198,14 @@ def insert_upsert_implementation(
                 )
             else:
                 raise
-        if tracker is not None and db.table(table).exists():
+        # Apply detected types only to a table this command created -
+        # transforming a pre-existing table would rewrite its column types
+        # and corrupt values such as TEXT zip codes with leading zeros
+        if (
+            tracker is not None
+            and not table_existed_before_insert
+            and db.table(table).exists()
+        ):
             db.table(table).transform(types=tracker.types)
 
         # Clean up open file-like objects
@@ -2724,7 +2735,10 @@ def extract(
         fk_column=fk_column,
         rename=dict(rename),
     )
-    db.table(table).extract(**kwargs)
+    try:
+        db.table(table).extract(**kwargs)
+    except (NoTable, InvalidColumns) as e:
+        raise click.ClickException(str(e))
 
 
 @cli.command(name="insert-files")
@@ -3420,7 +3434,14 @@ def migrate(db_path, migrations, stop_before, list_, verbose):
             # Listing is read-only - don't create the database file
             db = sqlite_utils.Database(memory=True)
         _register_db_for_cleanup(db)
-        _display_migration_list(db, migration_sets)
+        # Legacy sqlite-migrate classes create the migrations table from
+        # their pending()/applied() methods - run the listing inside a
+        # transaction and roll it back so --list stays read-only
+        db.begin()
+        try:
+            _display_migration_list(db, migration_sets)
+        finally:
+            db.rollback()
         return
 
     db = sqlite_utils.Database(db_path)
@@ -3452,7 +3473,10 @@ def migrate(db_path, migrations, stop_before, list_, verbose):
     for migration_set in migration_sets:
         matches = _stop_before_for_migration_set(stop_before, migration_set.name)
         if isinstance(migration_set, sqlite_utils.Migrations):
-            migration_set.apply(db, stop_before=matches)
+            try:
+                migration_set.apply(db, stop_before=matches)
+            except ValueError as e:
+                raise click.ClickException(str(e))
         else:
             # Legacy sqlite-migrate Migrations objects take a single string
             # for stop_before, not a list
diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py
index bd023c577..bec68fc89 100644
--- a/sqlite_utils/db.py
+++ b/sqlite_utils/db.py
@@ -193,7 +193,7 @@ def resolve_casing(name: str, candidates: Iterable[str]) -> str:
 """
 
 
-@dataclass(order=True)
+@dataclass(order=True, frozen=True)
 class ForeignKey:
     """
     A foreign key defined on a table.
@@ -208,6 +208,11 @@ class ForeignKey:
     ``on_delete`` and ``on_update`` hold the foreign key actions, e.g.
     ``"CASCADE"`` - ``"NO ACTION"`` if not set.
 
+    Instances are immutable and hashable, so they can be collected into
+    sets and used as dictionary keys. Equality covers every compared field,
+    including ``on_delete`` and ``on_update`` - two foreign keys differing
+    only in their actions are different constraints.
+
     Prior to sqlite-utils 4.0 this was a ``namedtuple`` and could be unpacked
     or indexed as ``(table, column, other_table, other_column)``. It is now a
     dataclass - access its fields by name instead.
@@ -227,16 +232,21 @@ class ForeignKey:
 
     def __post_init__(self):
         # Populate columns/other_columns for single-column foreign keys,
-        # normalizing any lists to tuples
+        # normalizing any lists to tuples. object.__setattr__ because the
+        # dataclass is frozen
         if self.columns:
-            self.columns = tuple(self.columns)
+            object.__setattr__(self, "columns", tuple(self.columns))
         else:
-            self.columns = (self.column,) if self.column is not None else ()
+            object.__setattr__(
+                self, "columns", (self.column,) if self.column is not None else ()
+            )
         if self.other_columns:
-            self.other_columns = tuple(self.other_columns)
+            object.__setattr__(self, "other_columns", tuple(self.other_columns))
         else:
-            self.other_columns = (
-                (self.other_column,) if self.other_column is not None else ()
+            object.__setattr__(
+                self,
+                "other_columns",
+                (self.other_column,) if self.other_column is not None else (),
             )
 
 
@@ -592,8 +602,13 @@ def atomic(self) -> Generator["Database", None, None]:
             try:
                 yield self
             except BaseException:
-                self.conn.execute("ROLLBACK TO SAVEPOINT {};".format(savepoint))
-                self.conn.execute("RELEASE SAVEPOINT {};".format(savepoint))
+                # An error such as a RAISE(ROLLBACK) trigger can destroy
+                # the whole transaction, savepoints included - cleaning up
+                # anyway would mask the original exception with
+                # "no such savepoint"
+                if self.conn.in_transaction:
+                    self.conn.execute("ROLLBACK TO SAVEPOINT {};".format(savepoint))
+                    self.conn.execute("RELEASE SAVEPOINT {};".format(savepoint))
                 raise
             else:
                 self.conn.execute("RELEASE SAVEPOINT {};".format(savepoint))
@@ -602,13 +617,15 @@ def atomic(self) -> Generator["Database", None, None]:
             try:
                 yield self
             except BaseException:
-                self.conn.execute("ROLLBACK")
+                # rollback() is a no-op if the error already destroyed the
+                # transaction, so the original exception propagates
+                self.rollback()
                 raise
             else:
                 try:
                     self.conn.execute("COMMIT")
                 except BaseException:
-                    self.conn.execute("ROLLBACK")
+                    self.rollback()
                     raise
 
     def begin(self) -> None:
@@ -655,7 +672,16 @@ def ensure_autocommit_on(self) -> Generator[None, None, None]:
                 # do stuff here
 
         The previous ``isolation_level`` is restored at the end of the block.
+
+        :raises TransactionError: if a transaction is open - assigning
+          ``isolation_level`` would commit it as a side effect, silently
+          breaking the caller's ability to roll back
         """
+        if self.conn.in_transaction:
+            raise TransactionError(
+                "ensure_autocommit_on() cannot be used inside a transaction - "
+                "changing isolation_level would commit the open transaction"
+            )
         old_isolation_level = self.conn.isolation_level
         try:
             self.conn.isolation_level = None
@@ -797,7 +823,10 @@ def query(
           parameters, or a dictionary for ``where id = :id``
         :raises ValueError: if the SQL statement does not return rows - use
           :meth:`execute` for those statements instead. The rejected statement
-          is rolled back, so it has no effect on the database
+          is rolled back, so it has no effect on the database. One exception:
+          a row-less ``PRAGMA`` statement takes effect despite the
+          ``ValueError``, because PRAGMAs run outside the savepoint guard -
+          some of them refuse to run inside a transaction
         """
         message = (
             "query() can only be used with SQL that returns rows - "
@@ -848,8 +877,11 @@ def query(
                 return (dict(zip(keys, row)) for row in fetched)
             return (dict(zip(keys, row)) for row in cursor)
         finally:
-            if not released:
-                # An error occurred - undo anything the statement changed
+            if not released and self.conn.in_transaction:
+                # An error occurred - undo anything the statement changed.
+                # If the error itself destroyed the transaction (such as a
+                # RAISE(ROLLBACK) trigger) the savepoint is already gone
+                # and there is nothing left to undo
                 self.conn.execute('ROLLBACK TO "sqlite_utils_query"')
                 self.conn.execute('RELEASE "sqlite_utils_query"')
 
@@ -1225,21 +1257,23 @@ def resolve_foreign_keys(
             (("campus_name", "dept_code"), "departments", ("campus_name", "dept_code"))
         """
         table = self.table(name)
-        if all(isinstance(fk, ForeignKey) for fk in foreign_keys):
-            return cast(List[ForeignKey], foreign_keys)
-        if all(isinstance(fk, str) for fk in foreign_keys):
-            # It's a list of columns
-            fks = []
-            for column in foreign_keys:
-                column = cast(str, column)
-                other_table = table.guess_foreign_table(column)
-                other_column = table.guess_foreign_column(other_table)
-                fks.append(ForeignKey(name, column, other_table, other_column))
-            return fks
-        if not all(isinstance(fk, (tuple, list)) for fk in foreign_keys):
-            raise ValueError("foreign_keys= should be a list of tuples")
         fks = []
-        for tuple_or_list in cast(Iterable[Sequence[Any]], foreign_keys):
+        for fk in foreign_keys:
+            if isinstance(fk, ForeignKey):
+                fks.append(fk)
+                continue
+            if isinstance(fk, str):
+                # A bare column name - guess the other table and column
+                other_table = table.guess_foreign_table(fk)
+                other_column = table.guess_foreign_column(other_table)
+                fks.append(ForeignKey(name, fk, other_table, other_column))
+                continue
+            if not isinstance(fk, (tuple, list)):
+                raise ValueError(
+                    "foreign_keys= should be a list of tuples, "
+                    "ForeignKey objects or column name strings"
+                )
+            tuple_or_list = cast(Sequence[Any], fk)
             if len(tuple_or_list) == 4:
                 if tuple_or_list[0] != name:
                     raise ValueError(
@@ -1762,6 +1796,11 @@ def add_foreign_keys(
                     if isinstance(other_column_or_columns, str)
                     else tuple(other_column_or_columns)
                 )
+                if len(columns) != len(other_columns):
+                    raise ValueError(
+                        "Compound foreign key must have the same number of "
+                        "columns on both sides"
+                    )
                 if len(columns) == 1:
                     fk_object = ForeignKey(
                         table, columns[0], other_table, other_columns[0]
@@ -1801,10 +1840,11 @@ def add_foreign_keys(
                             other_column, other_table
                         )
                     )
-            # We will silently skip foreign keys that exist already
+            # Silently skip foreign keys that exist already - but only if
+            # they match exactly, including ON DELETE/ON UPDATE actions
             columns_folded = tuple(fold_identifier_case(c) for c in columns)
             other_columns_folded = tuple(fold_identifier_case(c) for c in other_columns)
-            if not any(
+            existing = [
                 fk
                 for fk in table_obj.foreign_keys
                 if tuple(fold_identifier_case(c) for c in fk.columns) == columns_folded
@@ -1812,8 +1852,21 @@ def add_foreign_keys(
                 == fold_identifier_case(other_table)
                 and tuple(fold_identifier_case(c) for c in fk.other_columns)
                 == other_columns_folded
-            ):
+            ]
+            if not existing:
                 foreign_keys_to_create.append(fk_object)
+            elif any(
+                fk.on_delete != fk_object.on_delete
+                or fk.on_update != fk_object.on_update
+                for fk in existing
+            ):
+                raise AlterError(
+                    "Foreign key already exists for {} => {}.{} but with "
+                    "different ON DELETE/ON UPDATE actions - use "
+                    "table.transform() to change them".format(
+                        ", ".join(columns), other_table, ", ".join(other_columns)
+                    )
+                )
 
         # Group them by table
         by_table: Dict[str, List[ForeignKey]] = {}
@@ -2008,12 +2061,22 @@ def pks_and_rows_where(
         :param limit: Integer number of rows to limit to
         :param offset: Integer for SQL offset
         """
-        column_names = [column.name for column in self.columns]
-        pks = [column.name for column in self.columns if column.is_pk]
+        # This method is defined on Queryable so it serves views too, which
+        # have no pks property - sort pk columns into declaration order here
+        pk_columns = sorted(
+            (column for column in self.columns if column.is_pk),
+            key=lambda column: column.is_pk,
+        )
+        pks = [column.name for column in pk_columns]
+        select_parts = [quote_identifier(column.name) for column in self.columns]
         if not pks:
-            column_names.insert(0, "rowid")
+            # rowid is left unquoted: it is not a real column, and SQLite
+            # turns a double-quoted identifier that does not resolve into a
+            # string literal - on a view that would silently select the
+            # string 'rowid' instead of raising an error
+            select_parts.insert(0, "rowid")
             pks = ["rowid"]
-        select = ",".join(quote_identifier(column_name) for column_name in column_names)
+        select = ",".join(select_parts)
         for row in self.rows_where(
             select=select,
             where=where,
@@ -2145,8 +2208,18 @@ def exists(self) -> bool:
 
     @property
     def pks(self) -> List[str]:
-        "Primary key columns for this table."
-        names = [column.name for column in self.columns if column.is_pk]
+        """
+        Primary key columns for this table, in PRIMARY KEY declaration order -
+        ``PRAGMA table_info`` sets ``is_pk`` to the 1-based position of each
+        column within the primary key, which can differ from the order of the
+        columns in the table. SQLite uses the declaration order to resolve
+        implicit foreign key references, so this order matters.
+        """
+        pk_columns = sorted(
+            (column for column in self.columns if column.is_pk),
+            key=lambda column: column.is_pk,
+        )
+        names = [column.name for column in pk_columns]
         if not names:
             names = ["rowid"]
         return names
@@ -2670,7 +2743,8 @@ def fk_with_renamed_columns(fk: ForeignKey) -> ForeignKey:
 
         if pk is DEFAULT:
             pks_renamed = tuple(
-                rename.get(p.name) or p.name for p in self.columns if p.is_pk
+                rename.get(pk_name) or pk_name
+                for pk_name in (self.pks if not self.use_rowid else [])
             )
             if len(pks_renamed) == 1:
                 pk = pks_renamed[0]
@@ -2855,8 +2929,22 @@ def extract(
             all_columns_are_null = " AND ".join(
                 "{} IS NULL".format(quote_identifier(c)) for c in columns
             )
+            # INSERT OR IGNORE dedupes against the unique index, but unique
+            # indexes treat NULLs as distinct - the NOT EXISTS guard uses IS
+            # comparison so NULL-containing rows match existing lookup rows
+            # instead of being inserted again
+            already_in_lookup = " AND ".join(
+                "{lookup}.{lookup_col} IS {source}.{source_col}".format(
+                    lookup=quote_identifier(table),
+                    lookup_col=quote_identifier(rename.get(column) or column),
+                    source=quote_identifier(self.name),
+                    source_col=quote_identifier(column),
+                )
+                for column in columns
+            )
             self.db.execute(
-                "INSERT OR IGNORE INTO {} ({lookup_columns}) SELECT DISTINCT {table_cols} FROM {} WHERE NOT ({all_null})".format(
+                "INSERT OR IGNORE INTO {} ({lookup_columns}) SELECT DISTINCT {table_cols} FROM {} "
+                "WHERE NOT ({all_null}) AND NOT EXISTS (SELECT 1 FROM {lookup} WHERE {already_in_lookup})".format(
                     quote_identifier(table),
                     quote_identifier(self.name),
                     lookup_columns=", ".join(
@@ -2864,6 +2952,8 @@ def extract(
                     ),
                     table_cols=", ".join(quote_identifier(c) for c in columns),
                     all_null=all_columns_are_null,
+                    lookup=quote_identifier(table),
+                    already_in_lookup=already_in_lookup,
                 )
             )
 
@@ -3014,7 +3104,10 @@ def add_column(
                     raise AlterError("table '{}' has no column {}".format(fk, fk_col))
             else:
                 # automatically set fk_col to first primary_key of fk table
-                pks = [c for c in self.db[fk].columns if c.is_pk]
+                pks = sorted(
+                    (c for c in self.db[fk].columns if c.is_pk),
+                    key=lambda c: c.is_pk,
+                )
                 if pks:
                     fk_col = pks[0].name
                     fk_col_type = pks[0].type
@@ -3767,9 +3860,7 @@ def _convert_multi(
         # First we execute the function
         pk_to_values = {}
         new_column_types: Dict[str, Set[type]] = {}
-        pks = [column.name for column in self.columns if column.is_pk]
-        if not pks:
-            pks = ["rowid"]
+        pks = self.pks
 
         with progressbar(
             length=self.count, silent=not show_progress, label="1: Evaluating"
@@ -4245,6 +4336,10 @@ def insert_all(
         if hash_id:
             pk = hash_id
 
+        # pk columns missing from an existing table are an error - unless
+        # alter=True, where a pk column supplied by the records will be
+        # added, so validation waits until the record keys are known
+        deferred_invalid_pk_check = None
         if pk and not hash_id and self.exists():
             pk_cols = [pk] if isinstance(pk, str) else list(pk)
             existing_columns = self.columns_dict
@@ -4254,7 +4349,7 @@ def insert_all(
                 if resolve_casing(col, existing_columns) not in existing_columns
             ]
             if missing_pk_cols:
-                raise InvalidColumns(
+                invalid_pk_error = InvalidColumns(
                     "Invalid primary key column{} {} for table {} with columns {}".format(
                         "s" if len(missing_pk_cols) > 1 else "",
                         missing_pk_cols,
@@ -4262,6 +4357,9 @@ def insert_all(
                         list(existing_columns),
                     )
                 )
+                if not alter:
+                    raise invalid_pk_error
+                deferred_invalid_pk_check = (missing_pk_cols, invalid_pk_error)
 
         if ignore and replace:
             raise ValueError("Use either ignore=True or replace=True, not both")
@@ -4372,6 +4470,16 @@ def insert_all(
                     all_columns = list(sorted(all_columns_set))
                     if hash_id:
                         all_columns.insert(0, hash_id)
+                if deferred_invalid_pk_check is not None:
+                    # alter=True - pk columns the table lacks are valid if
+                    # the records supply them, otherwise raise the error
+                    missing_pk_cols, invalid_pk_error = deferred_invalid_pk_check
+                    record_columns = {column: True for column in all_columns}
+                    if any(
+                        resolve_casing(col, record_columns) not in record_columns
+                        for col in missing_pk_cols
+                    ):
+                        raise invalid_pk_error
             else:
                 if not list_mode:
                     for record in cast(List[Dict[str, Any]], chunk):
diff --git a/sqlite_utils/migrations.py b/sqlite_utils/migrations.py
index 36e053d51..00d0fa5b0 100644
--- a/sqlite_utils/migrations.py
+++ b/sqlite_utils/migrations.py
@@ -96,14 +96,34 @@ def apply(self, db: "Database", *, stop_before: str | Iterable[str] | None = Non
         changes are rolled back, no record is written and the migration stays
         pending. Migrations registered with ``transactional=False`` run
         outside of a transaction.
+
+        :raises ValueError: if a ``stop_before`` name matches a migration in
+          this set that has already been applied - stopping before it is
+          impossible to honor, and no pending migrations are applied
         """
-        self.ensure_migrations_table(db)
         if stop_before is None:
             stop_before_names = set()
         elif isinstance(stop_before, str):
             stop_before_names = {stop_before}
         else:
             stop_before_names = set(stop_before)
+        # A stop_before naming an already-applied migration cannot be
+        # honored - error rather than applying everything after it. Names
+        # not in this set at all are ignored, because unqualified CLI
+        # values are offered to every migration set
+        already_applied = stop_before_names.intersection(
+            migration.name for migration in self.applied(db)
+        )
+        if already_applied:
+            raise ValueError(
+                "Cannot stop before migration{} {} in set '{}' - already "
+                "been applied".format(
+                    "s" if len(already_applied) > 1 else "",
+                    ", ".join(sorted(already_applied)),
+                    self.name,
+                )
+            )
+        self.ensure_migrations_table(db)
         for migration in self.pending(db):
             name = migration.name
             if name in stop_before_names:
diff --git a/tests/test_atomic.py b/tests/test_atomic.py
index 0d25b8457..c3fd02f6c 100644
--- a/tests/test_atomic.py
+++ b/tests/test_atomic.py
@@ -258,11 +258,21 @@ def test_execute_comment_prefixed_begin_leaves_transaction_open(fresh_db):
     assert [r["id"] for r in fresh_db["t"].rows] == [1]
 
 
+def _sqlite_accepts_bom():
+    try:
+        sqlite3.connect(":memory:").execute("\ufeffselect 1")
+        return True
+    except sqlite3.OperationalError:
+        return False
+
+
 @pytest.mark.parametrize("begin_sql", ["; begin", "\ufeffbegin"])
 def test_execute_prefixed_begin_leaves_transaction_open(fresh_db, begin_sql):
     # sqlite3 tolerates empty statements and a UTF-8 BOM before the first
     # real token, so a BEGIN behind either must not be auto-committed
     # out from under the caller
+    if begin_sql.startswith("\ufeff") and not _sqlite_accepts_bom():
+        pytest.skip("This SQLite version rejects a leading byte order mark")
     fresh_db["t"].insert({"id": 1}, pk="id")
     fresh_db.execute(begin_sql)
     assert fresh_db.conn.in_transaction
@@ -327,3 +337,46 @@ def test_query_returning_commits_after_iteration(tmpdir):
     assert other.execute("select count(*) from t").fetchone()[0] == 2
     other.close()
     db.close()
+
+
+TRIGGER_SQL = """
+create trigger no_bad before insert on t
+when new.v = 'bad'
+begin
+    select raise(rollback, 'trigger says no');
+end
+"""
+
+
+def test_atomic_preserves_error_from_transaction_destroying_trigger(fresh_db):
+    # RAISE(ROLLBACK) rolls back the whole transaction and destroys every
+    # savepoint - atomic()'s cleanup must not mask the IntegrityError
+    # with "cannot rollback - no transaction is active"
+    fresh_db.execute("create table t (id integer primary key, v text)")
+    fresh_db.execute(TRIGGER_SQL)
+    with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
+        with fresh_db.atomic():
+            fresh_db.execute("insert into t (v) values ('bad')")
+    assert not fresh_db.conn.in_transaction
+
+
+def test_nested_atomic_preserves_error_from_transaction_destroying_trigger(
+    fresh_db,
+):
+    # The nested savepoint branch previously raised
+    # "no such savepoint" from ROLLBACK TO SAVEPOINT
+    fresh_db.execute("create table t (id integer primary key, v text)")
+    fresh_db.execute(TRIGGER_SQL)
+    with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
+        with fresh_db.atomic():
+            with fresh_db.atomic():
+                fresh_db.execute("insert into t (v) values ('bad')")
+    assert not fresh_db.conn.in_transaction
+
+
+def test_atomic_preserves_error_from_insert_or_rollback(fresh_db):
+    fresh_db["t"].insert({"id": 1}, pk="id")
+    with pytest.raises(sqlite3.IntegrityError):
+        with fresh_db.atomic():
+            fresh_db.execute("insert or rollback into t (id) values (1)")
+    assert not fresh_db.conn.in_transaction
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 9e1796953..06e14ea8d 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -2758,3 +2758,22 @@ def test_insert_upsert_strict(tmpdir, method, strict):
     assert result.exit_code == 0
     db = Database(db_path)
     assert db["items"].strict == strict or not db.supports_strict
+
+
+def test_extract_bad_column_clean_error(db_path):
+    db = Database(db_path)
+    db["trees"].insert({"id": 1, "species": "Palm"}, pk="id")
+    result = CliRunner().invoke(cli.cli, ["extract", db_path, "trees", "nope"])
+    assert result.exit_code == 1
+    assert result.exception is None or isinstance(result.exception, SystemExit)
+    assert result.output.startswith("Error: Invalid columns")
+
+
+def test_extract_view_clean_error(db_path):
+    db = Database(db_path)
+    db["trees"].insert({"id": 1, "species": "Palm"}, pk="id")
+    db.create_view("v", "select * from trees")
+    result = CliRunner().invoke(cli.cli, ["extract", db_path, "v", "species"])
+    assert result.exit_code == 1
+    assert result.exception is None or isinstance(result.exception, SystemExit)
+    assert result.output.startswith("Error:")
diff --git a/tests/test_cli_insert.py b/tests/test_cli_insert.py
index 2df1e0ceb..5af8a2fa3 100644
--- a/tests/test_cli_insert.py
+++ b/tests/test_cli_insert.py
@@ -628,3 +628,66 @@ def test_insert_into_view_errors(tmpdir):
     )
     assert result.exit_code == 1
     assert result.output.strip() == "Error: Table v is actually a view"
+
+
+def test_insert_csv_detect_types_leaves_existing_table_alone(db_path):
+    # Type detection is the default for CSV/TSV inserts, but it must only
+    # apply to tables created by this command - transforming a pre-existing
+    # table would rewrite its column types and corrupt data such as
+    # TEXT zip codes with leading zeros
+    db = Database(db_path)
+    db["places"].insert({"name": "Boston", "zip": "01234"})
+    result = CliRunner().invoke(
+        cli.cli,
+        ["insert", db_path, "places", "-", "--csv"],
+        catch_exceptions=False,
+        input="name,zip\nSF,94107",
+    )
+    assert result.exit_code == 0, result.output
+    assert db["places"].columns_dict["zip"] is str
+    assert list(db["places"].rows) == [
+        {"name": "Boston", "zip": "01234"},
+        {"name": "SF", "zip": "94107"},
+    ]
+
+
+def test_insert_csv_detect_types_new_table(db_path):
+    # A table created by the insert still gets detected types
+    result = CliRunner().invoke(
+        cli.cli,
+        ["insert", db_path, "data", "-", "--csv"],
+        catch_exceptions=False,
+        input="name,age,weight\nCleo,5,12.5",
+    )
+    assert result.exit_code == 0, result.output
+    db = Database(db_path)
+    assert db["data"].columns_dict == {"name": str, "age": int, "weight": float}
+
+
+def test_upsert_csv_detect_types_leaves_existing_table_alone(db_path):
+    db = Database(db_path)
+    db["places"].insert({"id": 1, "name": "Boston", "zip": "01234"}, pk="id")
+    result = CliRunner().invoke(
+        cli.cli,
+        ["upsert", db_path, "places", "-", "--csv", "--pk", "id"],
+        catch_exceptions=False,
+        input="id,name,zip\n2,SF,94107",
+    )
+    assert result.exit_code == 0, result.output
+    assert db["places"].columns_dict["zip"] is str
+    assert db["places"].get(1)["zip"] == "01234"
+
+
+def test_insert_invalid_pk_clean_error(db_path):
+    # An invalid --pk against an existing table should be a clean CLI
+    # error, not a raw InvalidColumns traceback
+    db = Database(db_path)
+    db["t"].insert({"a": 1})
+    result = CliRunner().invoke(
+        cli.cli,
+        ["insert", db_path, "t", "-", "--pk", "badcol"],
+        input='{"a": 2}',
+    )
+    assert result.exit_code == 1
+    assert result.exception is None or isinstance(result.exception, SystemExit)
+    assert result.output.startswith("Error: Invalid primary key column")
diff --git a/tests/test_cli_migrate.py b/tests/test_cli_migrate.py
index 1cdf8e7c7..0f29e36e1 100644
--- a/tests/test_cli_migrate.py
+++ b/tests/test_cli_migrate.py
@@ -463,3 +463,45 @@ def test_list_does_not_upgrade_legacy_migrations_table(two_migrations):
     db2 = sqlite_utils.Database(db_path)
     assert db2["_sqlite_migrations"].pks == ["migration_set", "name"]
     db2.close()
+
+
+def test_stop_before_applied_migration_errors(two_migrations):
+    path, _ = two_migrations
+    db_path = str(path / "test.db")
+    migrations_path = str(path / "foo" / "migrations.py")
+    # Apply everything first
+    first = CliRunner().invoke(
+        sqlite_utils.cli.cli,
+        ["migrate", db_path, migrations_path, "--stop-before", "bar"],
+    )
+    assert first.exit_code == 0
+    # foo is now applied - stopping before it is an error, and bar
+    # must not be applied as a side effect
+    result = CliRunner().invoke(
+        sqlite_utils.cli.cli,
+        ["migrate", db_path, migrations_path, "--stop-before", "foo"],
+    )
+    assert result.exit_code != 0
+    assert "already been applied" in result.output
+    db = sqlite_utils.Database(db_path)
+    assert not db["bar"].exists()
+
+
+def test_list_with_legacy_class_is_read_only(tmpdir):
+    # Legacy sqlite-migrate classes create the _sqlite_migrations table
+    # from their pending()/applied() methods - --list must roll that
+    # back so it stays a read-only operation as documented
+    path = pathlib.Path(tmpdir)
+    (path / "migrations.py").write_text(LEGACY_MIGRATIONS, "utf-8")
+    db_path = str(path / "test.db")
+    db = sqlite_utils.Database(db_path)
+    db["existing"].insert({"id": 1})
+    db.close()
+    result = CliRunner().invoke(
+        sqlite_utils.cli.cli, ["migrate", db_path, str(path), "--list"]
+    )
+    assert result.exit_code == 0, result.output
+    assert "first" in result.output
+    db2 = sqlite_utils.Database(db_path)
+    assert "_sqlite_migrations" not in db2.table_names()
+    db2.close()
diff --git a/tests/test_create.py b/tests/test_create.py
index 278892045..42fa1c422 100644
--- a/tests/test_create.py
+++ b/tests/test_create.py
@@ -932,6 +932,23 @@ def test_insert_all_pk_not_in_records_raises(fresh_db, num_rows):
     fresh_db.conn.execute("CREATE TABLE t (a TEXT, b INT, PRIMARY KEY (a, b))")
     rows = [{"a": "x{}".format(i), "b": i} for i in range(num_rows)]
 
+    with pytest.raises(InvalidColumns) as ex:
+        fresh_db["t"].insert_all(rows, pk="not_a_column")
+
+    assert ex.value.args == (
+        "Invalid primary key column ['not_a_column'] for table t with columns ['a', 'b']",
+    )
+    assert fresh_db["t"].count == 0
+
+
+@pytest.mark.parametrize("num_rows", (1, 2, 3, 10))
+def test_insert_all_pk_not_in_records_alter_raises(fresh_db, num_rows):
+    # With alter=True the check is deferred until the record keys are
+    # known - a pk column that is in neither the table nor the records
+    # still raises
+    fresh_db.conn.execute("CREATE TABLE t (a TEXT, b INT, PRIMARY KEY (a, b))")
+    rows = [{"a": "x{}".format(i), "b": i} for i in range(num_rows)]
+
     with pytest.raises(InvalidColumns) as ex:
         fresh_db["t"].insert_all(rows, pk="not_a_column", alter=True)
 
@@ -941,6 +958,26 @@ def test_insert_all_pk_not_in_records_raises(fresh_db, num_rows):
     assert fresh_db["t"].count == 0
 
 
+def test_insert_pk_in_records_with_alter_adds_column(fresh_db):
+    # 3.x allowed insert(pk=..., alter=True) to add the pk column from the
+    # records - the InvalidColumns check must not fire in that case
+    fresh_db["t"].insert({"a": 1})
+    fresh_db["t"].insert({"id": 5, "a": 2}, pk="id", alter=True)
+    assert fresh_db["t"].columns_dict.keys() == {"a", "id"}
+    assert list(fresh_db.query("select * from t order by a")) == [
+        {"a": 1, "id": None},
+        {"a": 2, "id": 5},
+    ]
+
+
+def test_insert_all_invalid_pk_alter_empty_records_is_noop(fresh_db):
+    # With alter=True the pk check needs record keys, so an empty iterator
+    # returns without error - matching the 3.x no-op for empty inserts
+    fresh_db.conn.execute("CREATE TABLE t (a TEXT)")
+    fresh_db["t"].insert_all([], pk="not_a_column", alter=True)
+    assert fresh_db["t"].count == 0
+
+
 def test_insert_ignore(fresh_db):
     fresh_db["test"].insert({"id": 1, "bar": 2}, pk="id")
     # Should raise an error if we try this again
diff --git a/tests/test_extract.py b/tests/test_extract.py
index 1c0fa01eb..c73ee7a62 100644
--- a/tests/test_extract.py
+++ b/tests/test_extract.py
@@ -271,3 +271,34 @@ def test_extract_null_values_existing_lookup_table_with_null_row(fresh_db):
         {"id": 10, "name": "Terriana", "species_id": 2},
         {"id": 11, "name": "Spenidorm", "species_id": None},
     ]
+
+
+def test_extract_repeated_into_shared_lookup_with_nulls(fresh_db):
+    # Unique indexes treat NULLs as distinct, so INSERT OR IGNORE alone
+    # cannot dedupe NULL-containing rows against the existing lookup
+    # table - extracting a second table into the same lookup previously
+    # inserted duplicate rows that nothing pointed to
+    fresh_db["t1"].insert_all(
+        [
+            {"id": 1, "species": None, "common": "X"},
+            {"id": 2, "species": "Oak", "common": "Oak"},
+        ],
+        pk="id",
+    )
+    fresh_db["t2"].insert_all([{"id": 1, "species": None, "common": "X"}], pk="id")
+    fresh_db["t1"].extract(["species", "common"], table="lk")
+    fresh_db["t2"].extract(["species", "common"], table="lk")
+    assert fresh_db["lk"].count == 2
+    # Both tables point at the same lookup row
+    t1_fk = fresh_db.execute("select lk_id from t1 where id = 1").fetchone()[0]
+    t2_fk = fresh_db.execute("select lk_id from t2 where id = 1").fetchone()[0]
+    assert t1_fk == t2_fk
+
+
+def test_extract_repeated_into_shared_lookup_no_nulls(fresh_db):
+    # Non-NULL rows were already deduped by the unique index - keep it so
+    fresh_db["t1"].insert_all([{"id": 1, "species": "Oak"}], pk="id")
+    fresh_db["t2"].insert_all([{"id": 1, "species": "Oak"}], pk="id")
+    fresh_db["t1"].extract(["species"], table="lk")
+    fresh_db["t2"].extract(["species"], table="lk")
+    assert fresh_db["lk"].count == 1
diff --git a/tests/test_foreign_keys.py b/tests/test_foreign_keys.py
index 8950189b0..b37d374a4 100644
--- a/tests/test_foreign_keys.py
+++ b/tests/test_foreign_keys.py
@@ -524,3 +524,168 @@ def test_add_compound_foreign_key_on_delete(courses_db):
     assert fk.is_compound is True
     assert fk.on_delete == "SET NULL"
     assert "ON DELETE SET NULL" in courses_db["courses"].schema
+
+
+def test_implicit_compound_foreign_key_resolves_pk_declaration_order(fresh_db):
+    # The other table's PRIMARY KEY declares its columns in a different
+    # order to the table's column order. SQLite resolves the implicit
+    # "REFERENCES other" using PRIMARY KEY declaration order, so the
+    # introspected other_columns must too
+    fresh_db.execute("create table other (b text, a text, primary key (a, b))")
+    fresh_db.execute(
+        "create table child (x text, y text, foreign key (x, y) references other)"
+    )
+    fk = fresh_db["child"].foreign_keys[0]
+    assert fk.other_columns == ("a", "b")
+
+
+def test_transform_implicit_compound_foreign_key_stays_valid(fresh_db):
+    # transform() rewrites the implicit FK with explicit columns - they
+    # must be in PRIMARY KEY declaration order or valid data fails the
+    # foreign key check with an IntegrityError
+    fresh_db.execute("create table other (b text, a text, primary key (a, b))")
+    fresh_db.execute(
+        "create table child (x text, y text, foreign key (x, y) references other)"
+    )
+    fresh_db.execute("PRAGMA foreign_keys = ON")
+    fresh_db["other"].insert({"a": "A", "b": "B"})
+    fresh_db["child"].insert({"x": "A", "y": "B"})
+    fresh_db["child"].transform(types={"x": str})
+    assert fresh_db["child"].foreign_keys[0].other_columns == ("a", "b")
+    # The constraint still points the right way around
+    fresh_db["child"].insert({"x": "A", "y": "B"})
+    with pytest.raises(sqlite3.IntegrityError):
+        fresh_db["child"].insert({"x": "B", "y": "A"})
+
+
+def test_create_compound_foreign_key_guesses_pk_declaration_order(fresh_db):
+    fresh_db.execute("create table other (b text, a text, primary key (a, b))")
+    fresh_db["other"].insert({"a": "A", "b": "B"})
+    fresh_db["child"].create(
+        {"id": int, "x": str, "y": str},
+        pk="id",
+        foreign_keys=[(("x", "y"), "other")],
+    )
+    assert fresh_db["child"].foreign_keys[0].other_columns == ("a", "b")
+    fresh_db.execute("PRAGMA foreign_keys = ON")
+    fresh_db["child"].insert({"id": 1, "x": "A", "y": "B"})
+    with pytest.raises(sqlite3.IntegrityError):
+        fresh_db["child"].insert({"id": 2, "x": "B", "y": "A"})
+
+
+def test_add_compound_foreign_key_guesses_pk_declaration_order(fresh_db):
+    fresh_db.execute("create table other (b text, a text, primary key (a, b))")
+    fresh_db["child"].insert({"id": 1, "x": "A", "y": "B"}, pk="id")
+    fresh_db["child"].add_foreign_key(("x", "y"), "other")
+    assert fresh_db["child"].foreign_keys[0].other_columns == ("a", "b")
+
+
+def test_foreign_keys_are_hashable(fresh_db):
+    # set() over foreign_keys worked with the 3.x namedtuple and must
+    # keep working with the dataclass
+    fresh_db["p"].insert({"id": 1}, pk="id")
+    fresh_db["c"].insert(
+        {"id": 1, "pid": 1}, pk="id", foreign_keys=[("pid", "p", "id")]
+    )
+    fks = set(fresh_db["c"].foreign_keys)
+    assert len(fks) == 1
+    assert ForeignKey("c", "pid", "p", "id") in fks
+    # Usable as dict keys too
+    assert {fk: True for fk in fks}
+
+
+def test_foreign_key_is_immutable():
+    import dataclasses
+
+    fk = ForeignKey("c", "pid", "p", "id")
+    with pytest.raises(dataclasses.FrozenInstanceError):
+        fk.table = "other"
+
+
+def test_foreign_key_equality_and_hash_include_actions():
+    # Two foreign keys differing only in ON DELETE behavior are different
+    # constraints - they compare unequal and hash separately
+    plain = ForeignKey("c", "pid", "p", "id")
+    cascade = ForeignKey("c", "pid", "p", "id", on_delete="CASCADE")
+    assert plain != cascade
+    assert len({plain, cascade}) == 2
+    assert plain == ForeignKey("c", "pid", "p", "id")
+
+
+def test_create_table_mixed_foreign_keys_list(fresh_db):
+    # 3.x accepted a mix of ForeignKey objects, tuples and bare column
+    # strings in foreign_keys= (ForeignKey was a namedtuple, so it passed
+    # the tuple check) - keep accepting the mix
+    fresh_db["authors"].insert({"id": 1}, pk="id")
+    fresh_db["publishers"].insert({"id": 1}, pk="id")
+    fresh_db["books"].create(
+        {"id": int, "author_id": int, "publisher_id": int},
+        pk="id",
+        foreign_keys=[
+            ForeignKey("books", "author_id", "authors", "id"),
+            ("publisher_id", "publishers", "id"),
+        ],
+    )
+    fks = {fk.column: fk.other_table for fk in fresh_db["books"].foreign_keys}
+    assert fks == {"author_id": "authors", "publisher_id": "publishers"}
+
+
+def test_create_table_mixed_foreign_keys_with_string(fresh_db):
+    fresh_db["authors"].insert({"id": 1}, pk="id")
+    fresh_db["publishers"].insert({"id": 1}, pk="id")
+    fresh_db["books"].create(
+        {"id": int, "author_id": int, "publisher_id": int},
+        pk="id",
+        foreign_keys=[
+            "author_id",  # bare column, table and column guessed
+            ("publisher_id", "publishers", "id"),
+        ],
+    )
+    fks = {fk.column: fk.other_table for fk in fresh_db["books"].foreign_keys}
+    assert fks == {"author_id": "authors", "publisher_id": "publishers"}
+
+
+def test_add_foreign_keys_existing_with_different_actions_errors(fresh_db):
+    # Requesting an existing foreign key with different ON DELETE/ON UPDATE
+    # actions was silently skipped, dropping the requested change
+    fresh_db["authors"].insert({"id": 1}, pk="id")
+    fresh_db["books"].insert(
+        {"id": 1, "author_id": 1},
+        pk="id",
+        foreign_keys=[("author_id", "authors", "id")],
+    )
+    with pytest.raises(AlterError) as ex:
+        fresh_db.add_foreign_keys(
+            [ForeignKey("books", "author_id", "authors", "id", on_delete="CASCADE")]
+        )
+    assert "ON DELETE" in str(ex.value)
+    assert fresh_db["books"].foreign_keys[0].on_delete == "NO ACTION"
+
+
+def test_add_foreign_keys_identical_existing_is_noop(fresh_db):
+    # An exact match, including actions, is silently skipped so repeated
+    # calls stay idempotent
+    fresh_db["authors"].insert({"id": 1}, pk="id")
+    fresh_db["books"].insert({"id": 1, "author_id": 1}, pk="id")
+    fresh_db["books"].add_foreign_key("author_id", "authors", "id", on_delete="CASCADE")
+    fresh_db.add_foreign_keys(
+        [ForeignKey("books", "author_id", "authors", "id", on_delete="CASCADE")]
+    )
+    fks = fresh_db["books"].foreign_keys
+    assert len(fks) == 1
+    assert fks[0].on_delete == "CASCADE"
+
+
+def test_add_foreign_keys_compound_column_count_mismatch_errors(fresh_db):
+    # Previously the extra other-column was silently discarded, creating
+    # a single-column foreign key to just ("id")
+    fresh_db["departments"].insert(
+        {"campus": "north", "code": "cs"}, pk=("campus", "code")
+    )
+    fresh_db["courses"].insert({"id": 1, "campus": "north"}, pk="id")
+    with pytest.raises(ValueError) as ex:
+        fresh_db.add_foreign_keys(
+            [("courses", ("campus",), "departments", ("campus", "code"))]
+        )
+    assert "same number of columns" in str(ex.value)
+    assert fresh_db["courses"].foreign_keys == []
diff --git a/tests/test_introspect.py b/tests/test_introspect.py
index 4ff3f7728..8b6765d86 100644
--- a/tests/test_introspect.py
+++ b/tests/test_introspect.py
@@ -321,3 +321,18 @@ def test_table_default_values(fresh_db, value):
     )
     default_values = fresh_db["default_values"].default_values
     assert default_values == {"value": value}
+
+
+def test_pks_use_primary_key_declaration_order(fresh_db):
+    # PRIMARY KEY (a, b) declared against columns stored in order (b, a) -
+    # pks must follow the declaration order, which is what SQLite uses to
+    # resolve implicit foreign key references and compound pk lookups
+    fresh_db.execute("create table t (b text, a text, primary key (a, b))")
+    assert fresh_db["t"].pks == ["a", "b"]
+
+
+def test_transform_preserves_compound_pk_declaration_order(fresh_db):
+    fresh_db.execute("create table t (a text, b text, c text, primary key (b, a))")
+    fresh_db["t"].transform(drop={"c"})
+    assert fresh_db["t"].pks == ["b", "a"]
+    assert 'PRIMARY KEY ("b", "a")' in fresh_db["t"].schema
diff --git a/tests/test_migrations.py b/tests/test_migrations.py
index 5634d7919..04185fc9e 100644
--- a/tests/test_migrations.py
+++ b/tests/test_migrations.py
@@ -214,3 +214,33 @@ def m001_again(db):
             pass
 
     assert "m001" in str(ex.value)
+
+
+def test_stop_before_applied_migration_errors(migrations):
+    # Stopping before a migration that has already been applied is
+    # impossible to honor - previously the stop name was only checked
+    # against pending migrations, so everything after it was applied
+    db = sqlite_utils.Database(memory=True)
+    migrations.apply(db, stop_before="m002")  # applies m001 only
+    with pytest.raises(ValueError) as ex:
+        migrations.apply(db, stop_before="m001")
+    assert "m001" in str(ex.value)
+    assert "already been applied" in str(ex.value)
+    # Nothing else was applied
+    assert not db["cats"].exists()
+
+
+def test_stop_before_applied_migration_errors_before_any_apply(migrations):
+    # The error fires before any pending migration runs, even those that
+    # come before the already-applied stop target in registration order
+    db = sqlite_utils.Database(memory=True)
+    only_second = Migrations("test")
+
+    @only_second()
+    def m002(db):
+        db["cats"].create({"name": str})
+
+    only_second.apply(db)  # m002 applied, m001 still pending
+    with pytest.raises(ValueError):
+        migrations.apply(db, stop_before="m002")
+    assert not db["dogs"].exists()
diff --git a/tests/test_query.py b/tests/test_query.py
index c2f67316f..06847da4c 100644
--- a/tests/test_query.py
+++ b/tests/test_query.py
@@ -122,6 +122,18 @@ def test_query_pragma(tmpdir):
     db.close()
 
 
+def test_query_rejected_pragma_still_takes_effect(fresh_db):
+    # Documented limitation: PRAGMAs run outside the savepoint guard,
+    # because some of them refuse to run inside a transaction - so a
+    # row-less PRAGMA takes effect even though it raises ValueError.
+    # If this test starts failing because the pragma was rolled back,
+    # the limitation has been fixed - update the docs in python-api.rst
+    # and the query() docstring to remove the carve-out
+    with pytest.raises(ValueError):
+        fresh_db.query("pragma user_version = 5")
+    assert fresh_db.execute("pragma user_version").fetchone()[0] == 5
+
+
 def test_query_comment_prefixed_pragma(tmpdir):
     from sqlite_utils import Database
 
@@ -268,3 +280,24 @@ def test_execute_returning_dicts(fresh_db):
     assert fresh_db.execute_returning_dicts("select * from test") == [
         {"id": 1, "bar": 2}
     ]
+
+
+@pytest.mark.skipif(
+    sqlite3.sqlite_version_info < (3, 35, 0),
+    reason="RETURNING requires SQLite 3.35.0 or higher",
+)
+def test_query_preserves_error_from_transaction_destroying_trigger(fresh_db):
+    # RAISE(ROLLBACK) destroys the savepoint guard - the original
+    # IntegrityError must propagate, not "no such savepoint"
+    fresh_db.execute("create table t (id integer primary key, v text)")
+    fresh_db.execute("""
+        create trigger no_bad before insert on t
+        when new.v = 'bad'
+        begin
+            select raise(rollback, 'trigger says no');
+        end
+    """)
+    with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
+        fresh_db.query("insert into t (id, v) values (1, 'bad') returning id")
+    assert not fresh_db.conn.in_transaction
+    assert fresh_db.execute("select count(*) from t").fetchone()[0] == 0
diff --git a/tests/test_rows.py b/tests/test_rows.py
index f050d5a24..46d4f5358 100644
--- a/tests/test_rows.py
+++ b/tests/test_rows.py
@@ -111,3 +111,30 @@ def test_rows_where_duplicate_select_columns_are_deduped(fresh_db):
     fresh_db["t"].insert({"id": 1, "name": "Cleo"})
     rows = list(fresh_db["t"].rows_where(select="id, id, name"))
     assert rows == [{"id": 1, "id_2": 1, "name": "Cleo"}]
+
+
+def test_pks_and_rows_where_view(fresh_db):
+    # pks_and_rows_where() lives on Queryable so views expose it, but
+    # SQLite views have no rowid. Modern SQLite (3.36+) raises an
+    # OperationalError from the generated SQL; older versions returned
+    # NULL for a view's rowid. Either way it must not fail earlier with
+    # an AttributeError from View lacking Table-only properties
+    from sqlite_utils.utils import sqlite3
+
+    fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
+    fresh_db.create_view("dog_names", "select name from dogs")
+    try:
+        result = list(fresh_db["dog_names"].pks_and_rows_where())
+    except sqlite3.OperationalError:
+        pass  # SQLite 3.36+: no such column: rowid
+    else:
+        # Older SQLite returns NULL rowids for views
+        assert result == [(None, {"rowid": None, "name": "Cleo"})]
+
+
+def test_pks_and_rows_where_compound_pk_declaration_order(fresh_db):
+    # Compound pks are returned in PRIMARY KEY declaration order
+    fresh_db.execute("create table t (b text, a text, primary key (a, b))")
+    fresh_db["t"].insert({"a": "A", "b": "B"})
+    pks_and_rows = list(fresh_db["t"].pks_and_rows_where())
+    assert pks_and_rows == [(("A", "B"), {"b": "B", "a": "A"})]
diff --git a/tests/test_wal.py b/tests/test_wal.py
index c5a9c6095..2ddcf5444 100644
--- a/tests/test_wal.py
+++ b/tests/test_wal.py
@@ -69,3 +69,20 @@ def test_enable_wal_noop_inside_transaction_is_allowed(db_path_tmpdir):
         db["test"].insert({"id": 1}, pk="id")
         db.enable_wal()
     assert [r["id"] for r in db["test"].rows] == [1]
+
+
+def test_ensure_autocommit_on_inside_transaction_raises(db_path_tmpdir):
+    # Setting isolation_level commits any pending transaction as a side
+    # effect, silently breaking the caller's rollback guarantee - so
+    # entering autocommit mode with a transaction open is an error
+    db, path, tmpdir = db_path_tmpdir
+    db["test"].insert({"id": 1}, pk="id")
+    db.begin()
+    db.execute("insert into test (id) values (2)")
+    with pytest.raises(TransactionError):
+        with db.ensure_autocommit_on():
+            pass
+    # The transaction is still open and can still be rolled back
+    assert db.conn.in_transaction
+    db.rollback()
+    assert [r["id"] for r in db["test"].rows] == [1]