@@ -115,6 +115,10 @@ def active_connection(database_url: str) -> dict[str, Any]:
                 raw_product_version = _required_text_probe(
                     connection, "select DOLT_VERSION()", "DOLT_VERSION()",
                 )
+                if raw_product_version.strip() != "2.2.2":
+                    raise UnsupportedOperationError(
+                        "The active Dolt product version must be exactly 2.2.2."
+                    )
                 identity_source = "SELECT @@version, @@version_comment, DOLT_VERSION()"
             elif "mariadb" in identity_text:
                 profile_name = "mariadb"
@@ -236,13 +240,14 @@ def _profile(
         "maximum_row_bytes": profile.max_row_bytes,
     }
     theoretical = (
-        {"maximum_value_bytes": profile.max_text_value_bytes}
+        {"maximum_value_bytes": 4_294_967_295}
         if name == "dolt" else declared
     )
     proposed = (
         {
             "maximum_physical_columns": declared["maximum_physical_columns"],
             "maximum_source_variables": declared["maximum_source_variables"],
+            "maximum_value_bytes": declared["maximum_value_bytes"],
             "maximum_row_bytes": declared["maximum_row_bytes"],
         }
         if name == "dolt" else None
@@ -264,7 +269,7 @@ def _profile(
                 "maximum_source_variables": "proposed Dolt adapter envelope",
                 "maximum_physical_columns": "proposed Dolt adapter envelope",
                 "identifier_limit": "observed on exact Dolt 2.2.2",
-                "maximum_value_bytes": "Dolt LONGTEXT theoretical type ceiling",
+                "maximum_value_bytes": "observed Dolt adapter value envelope",
                 "maximum_row_bytes": "proposed Dolt adapter envelope",
             }
             if name == "dolt"
@@ -302,7 +307,11 @@ def _profile(
             packet = int(observed["max_allowed_packet"])
             payload = max(0, (packet - 131_072) // 2)
             effective["maximum_value_bytes"] = min(
-                theoretical["maximum_value_bytes"], payload,
+                (
+                    declared["maximum_value_bytes"]
+                    if name == "dolt" else theoretical["maximum_value_bytes"]
+                ),
+                payload,
             )
             effective["maximum_statement_bytes"] = payload
             sources["maximum_value_bytes"] = "active @@max_allowed_packet worst-case payload"
@@ -329,6 +338,12 @@ def _profile(
         "effective_limits": effective,
         "effective_limits_status": status,
         "numeric_type": "DOUBLE PRECISION" if name == "postgresql" else "DOUBLE" if name in {"mysql", "mariadb", "dolt"} else "REAL",
+        "numeric_value_policy": {
+            "finite_binary64": "supported",
+            "nan": "rejected_before_ddl",
+            "positive_infinity": "rejected_before_ddl",
+            "negative_infinity": "rejected_before_ddl",
+        },
         "text_type": "LONGTEXT" if name in {"mysql", "mariadb", "dolt"} else "TEXT",
         "ddl_atomic": name not in {"mysql", "mariadb", "dolt"},
         "failure_cleanup": "compensating_cleanup" if name in {"mysql", "mariadb", "dolt"} else "transaction_rollback",
@@ -340,7 +355,9 @@ def _profile(
                 "proposed_adapter_envelope" if name == "dolt" else "theoretical_engine_limit"
             ),
             "identifier_limit": "observed_exact_version" if name == "dolt" else "theoretical_engine_limit",
-            "maximum_value_bytes": "theoretical_engine_limit",
+            "maximum_value_bytes": (
+                "observed_exact_version" if name == "dolt" else "theoretical_engine_limit"
+            ),
             "maximum_row_bytes": (
                 "proposed_adapter_envelope" if name == "dolt" else "theoretical_engine_limit"
             ),
@@ -373,11 +390,11 @@ def _profile(
 
 
 def server_version_supported(profile: str, raw_version: str) -> bool:
+    if profile == "dolt":
+        return raw_version.strip() == "2.2.2"
     version = _version_tuple(raw_version)
     if profile == "sqlite":
         return (3, 24) <= version[:2] < (4, 0)
-    if profile == "dolt":
-        return version == (2, 2, 2)
     allowed = {
         "postgresql": {(17,), (18,)},
         "mysql": {(8, 4), (9, 7)},
 
@@ -4,6 +4,8 @@
 Importers use this information for preflight checks before creating a dataset.
 """
 
+import math
+
 from collections.abc import Iterable, Mapping
 from dataclasses import dataclass
 from typing import Any
@@ -55,7 +57,7 @@ def as_dict(self) -> dict[str, object]:
 )
 DOLT = SqlProfile(
     "dolt", (), 305, 64, True, True,
-    4_294_967_295, 65_504, True, ("PyMySQL",),
+    65_504, 65_504, True, ("PyMySQL",),
 )
 # Dolt deliberately has no URL scheme. It is selected only after the server
 # reached through the MySQL wire family positively identifies itself as Dolt.
@@ -176,6 +178,15 @@ def preflight(
         row_bytes = 8
         for variable in variables:
             if variable.get("storage_kind") != "string":
+                value = row.get(str(variable["physical_name"]))
+                if value is not None and not math.isfinite(float(value)):
+                    raise _exceeded(
+                        "nonfinite_numeric_value",
+                        f"row {row_ordinal} value for {variable['source_name']!r} "
+                        "is not a finite binary64 value.",
+                        row_ordinal=row_ordinal,
+                        source_name=variable["source_name"],
+                    )
                 row_bytes += 8
                 continue
             value = row.get(str(variable["physical_name"]), "")
@@ -189,7 +200,7 @@ def preflight(
                     row_ordinal=row_ordinal, source_name=variable["source_name"],
                     encoded_bytes=encoded_bytes, maximum=profile.max_text_value_bytes,
                 )
-            row_bytes += 20 if profile.name in {"mysql", "mariadb"} else encoded_bytes
+            row_bytes += 20 if profile.name in {"mysql", "mariadb", "dolt"} else encoded_bytes
         if row_bytes > profile.max_row_bytes:
             raise _exceeded(
                 "row_size_limit",
@@ -203,7 +214,7 @@ def preflight(
 def _row_storage_bytes(profile: SqlProfile, variable: Mapping[str, Any]) -> int:
     if variable.get("storage_kind") == "numeric":
         return 8
-    if profile.name in {"mysql", "mariadb"}:
+    if profile.name in {"mysql", "mariadb", "dolt"}:
         return 20
     return int(variable.get("string_width") or 0)
 
 
@@ -209,6 +209,43 @@ def test_occupied_foreign_namespace_fails_without_modification(tmp_path) -> None
     assert connection.execute("select value from foreign_data").fetchall() == [("keep",)]
 
 
+@pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf")])
+def test_dolt_nonfinite_preflight_creates_no_dataset_or_physical_table(
+    tmp_path, monkeypatch, value,
+) -> None:
+    database_path = tmp_path / "dolt-nonfinite.sqlite"
+    database = f"sqlite:///{database_path}"
+    monkeypatch.setattr(wide, "effective_profile", lambda _url: (DOLT, {}))
+    variables = [{
+        "ordinal": 1, "source_name": "value", "physical_name": "value",
+        "storage_kind": "numeric", "string_width": None, "label": "",
+        "format": "F8.2", "measure": "scale", "alignment": "right",
+        "display_width": 8, "value_labels": "{}", "missing_ranges": "[]",
+    }]
+
+    with pytest.raises(Exception, match="Target capability exceeded"):
+        create_wide_dataset(
+            database_url=database, dataset_id="nonfinite",
+            source_name="nonfinite.sav", source_format="SAV",
+            rows=[{"value": value}], variables=variables,
+        )
+
+    connection = sqlite3.connect(database_path)
+    assert "data_nonfinite" not in {
+        row[0] for row in connection.execute(
+            "select name from sqlite_master where type = 'table'"
+        )
+    }
+    assert connection.execute("select count(*) from dataset").fetchone() == (0,)
+    assert connection.execute("select count(*) from variable").fetchone() == (0,)
+    assert connection.execute(
+        "select status from operation"
+    ).fetchall() == [("failed",)]
+    assert connection.execute(
+        "select dataset_id, event_code from fidelity_event"
+    ).fetchall() == [(None, "target_capability_exceeded")]
+
+
 def test_empty_namespace_dolt_width_failure_initializes_identity_and_one_audit(
     tmp_path, monkeypatch,
 ) -> None:
 
@@ -83,7 +83,9 @@ def test_normative_catalog_compiles_for_every_sql_family() -> None:
     [
         ("mysql", "8.4.6", True), ("mysql", "9.7.0", True),
         ("mysql", "8.0.44", False),
-        ("dolt", "2.2.2", True), ("dolt", "2.2.3", False),
+        ("dolt", "2.2.2", True), ("dolt", " 2.2.2 ", True),
+        ("dolt", "2.2.3", False), ("dolt", "2.2.2-rc1", False),
+        ("dolt", "2.2.2+build.1", False), ("dolt", "garbage", False),
         ("mariadb", "11.4.8-MariaDB", True),
         ("mariadb", "11.8.3-MariaDB", True),
         ("mariadb", "12.3.1-MariaDB", True),
@@ -223,6 +225,25 @@ def test_dolt_comment_requires_a_nonempty_product_version(monkeypatch) -> None:
         active_connection("mysql+pymysql://user@host/database")
 
 
+@pytest.mark.parametrize(
+    "product_version",
+    ["2.2.2-rc1", "2.2.2+build.1", "v2.2.2", "garbage"],
+)
+def test_dolt_identity_rejects_nonexact_product_versions(
+    monkeypatch, product_version,
+) -> None:
+    connection = _mock_mysql_probes(
+        monkeypatch, **{"select DOLT_VERSION()": product_version},
+    )
+
+    with pytest.raises(UnsupportedOperationError, match="exactly 2.2.2"):
+        active_connection("mysql+pymysql://user@host/database")
+
+    assert connection.calls == [
+        "select @@version", "select @@version_comment", "select DOLT_VERSION()",
+    ]
+
+
 def test_unknown_mysql_wire_product_fails_closed_without_dolt_probe(monkeypatch) -> None:
     connection = _mock_mysql_probes(
         monkeypatch,
@@ -268,6 +289,7 @@ def test_effective_profile_selects_dolt_without_changing_url_profile(monkeypatch
     assert profile.name == "dolt"
     assert profile.url_schemes == ()
     assert profile.max_physical_variables == 305
+    assert profile.max_text_value_bytes == 65_504
     assert profile.max_row_bytes == 65_504
     assert observed is active
 
@@ -281,22 +303,32 @@ def test_dolt_declaration_labels_conservative_envelopes() -> None:
     assert declaration["ci_tested_server_versions"] == ["Dolt 2.2.2"]
     assert declaration["proposed_adapter_limits"]["maximum_physical_columns"] == 306
     assert declaration["proposed_adapter_limits"]["maximum_source_variables"] == 305
+    assert declaration["proposed_adapter_limits"]["maximum_value_bytes"] == 65_504
     assert declaration["proposed_adapter_limits"]["maximum_row_bytes"] == 65_504
+    assert declaration["theoretical_limits"]["maximum_value_bytes"] == 4_294_967_295
     assert declaration["observed_limits"]["minimum_observed_physical_columns"] == 307
     assert declaration["observed_limits"]["identifier_limit"]["value"] == 64
     assert declaration["observed_limits"]["rejected_identifier_bytes"] == 65
     assert set(declaration["proposed_adapter_limits"]) == {
-        "maximum_physical_columns", "maximum_source_variables", "maximum_row_bytes",
+        "maximum_physical_columns", "maximum_source_variables",
+        "maximum_value_bytes", "maximum_row_bytes",
     }
     assert declaration["limit_bases"]["maximum_physical_columns"] == "proposed_adapter_envelope"
     assert declaration["limit_bases"]["identifier_limit"] == "observed_exact_version"
+    assert declaration["limit_bases"]["maximum_value_bytes"] == "observed_exact_version"
     assert declaration["limit_bases"]["maximum_statement_bytes"] == "active_connection_observation"
     assert declaration["effective_limits"] is None
     assert declaration["text_type"] == "LONGTEXT"
     assert capabilities.profile_declarations()["mysql"]["text_type"] == "LONGTEXT"
     assert capabilities.profile_declarations()["mariadb"]["text_type"] == "LONGTEXT"
     assert declaration["ddl_atomic"] is False
     assert declaration["failure_cleanup"] == "compensating_cleanup"
+    assert declaration["numeric_value_policy"] == {
+        "finite_binary64": "supported",
+        "nan": "rejected_before_ddl",
+        "positive_infinity": "rejected_before_ddl",
+        "negative_infinity": "rejected_before_ddl",
+    }
     assert declaration["storage_evidence"]["binary64"]["maximum_finite_round_trip_exact"] is True
     assert declaration["storage_evidence"]["binary64"]["source"]
     assert declaration["storage_evidence"]["binary64"]["version"] == "2.2.2"
@@ -320,13 +352,30 @@ def test_dolt_uses_longtext_without_changing_mysql_storage() -> None:
 def test_dolt_row_preflight_counts_utf8_values_against_adapter_envelope() -> None:
     variables = [{
         "ordinal": 1, "source_name": "value", "physical_name": "value",
-        "storage_kind": "string", "string_width": 65_496,
+        "storage_kind": "string", "string_width": 65_504,
     }]
-    preflight(DOLT, variables, rows=[{"value": "x" * 65_496}])
+    preflight(DOLT, variables, rows=[{"value": "x" * 65_504}])
 
     with pytest.raises(UnsupportedOperationError) as error:
-        preflight(DOLT, variables, rows=[{"value": "x" * 65_497}])
-    assert error.value.details["reason"] == "row_size_limit"
+        preflight(DOLT, variables, rows=[{"value": "x" * 65_505}])
+    assert error.value.details["reason"] == "text_value_limit"
+
+
+@pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf")])
+def test_dolt_numeric_preflight_rejects_nonfinite_values(value) -> None:
+    variables = [{
+        "ordinal": 1, "source_name": "value", "physical_name": "value",
+        "storage_kind": "numeric", "string_width": None,
+    }]
+
+    with pytest.raises(UnsupportedOperationError) as error:
+        preflight(DOLT, variables, rows=[{"value": value}])
+
+    assert error.value.details == {
+        "reason": "nonfinite_numeric_value",
+        "row_ordinal": 1,
+        "source_name": "value",
+    }
 
 
 def test_validate_identity_failure_happens_before_catalog_access(monkeypatch) -> None:
 
@@ -200,6 +200,57 @@ def fail_after_normative_write(*args, **kwargs):
     assert f"data_{dataset_id}" not in inspect_database(engine).get_table_names()
 
 
+def test_live_dolt_adapter_value_boundary_is_atomic() -> None:
+    database_url = os.environ.get("OPENSTATSPEC_DOLT_URL")
+    if not database_url:
+        pytest.skip("OPENSTATSPEC_DOLT_URL is not configured")
+    token = uuid4().hex[:8]
+    accepted_id = f"dolt_value_accepted_{token}"
+    rejected_id = f"dolt_value_rejected_{token}"
+    accepted_value = "é" * 32_752
+    rejected_value = accepted_value + "x"
+    assert len(accepted_value.encode("utf-8")) == 65_504
+    assert len(rejected_value.encode("utf-8")) == 65_505
+    variables = [{
+        "ordinal": 1, "source_name": "value", "physical_name": "value",
+        "storage_kind": "string", "string_width": 65_504, "label": "",
+        "format": "A65504", "measure": "nominal", "alignment": "left",
+        "display_width": 8, "value_labels": "{}", "missing_ranges": "[]",
+    }]
+
+    imported = wide.create_wide_dataset(
+        database_url=database_url, dataset_id=accepted_id,
+        source_name="accepted.sav", source_format="SAV",
+        rows=[{"value": accepted_value}], variables=variables,
+    )
+    assert imported["case_count"] == 1
+
+    with pytest.raises(UnsupportedOperationError) as caught:
+        wide.create_wide_dataset(
+            database_url=database_url, dataset_id=rejected_id,
+            source_name="rejected.sav", source_format="SAV",
+            rows=[{"value": rejected_value}], variables=variables,
+        )
+    assert caught.value.details["reason"] == "text_value_limit"
+
+    engine = create_engine(database_url)
+    accepted_table = wide.data_table_name(accepted_id)
+    rejected_table = wide.data_table_name(rejected_id)
+    quote = engine.dialect.identifier_preparer.quote
+    with engine.connect() as connection:
+        assert connection.execute(text(
+            f"SELECT OCTET_LENGTH(value) FROM {quote(accepted_table)}"
+        )).scalar_one() == 65_504
+        assert connection.execute(text(
+            "SELECT COUNT(*) FROM dataset_catalog WHERE dataset_id = :dataset_id"
+        ), {"dataset_id": rejected_id}).scalar_one() == 0
+        assert connection.execute(text(
+            "SELECT COUNT(*) FROM dataset WHERE dataset_name = :dataset_id"
+        ), {"dataset_id": rejected_id}).scalar_one() == 0
+    assert rejected_table not in inspect_database(engine).get_table_names()
+    engine.dispose()
+
+
 def test_live_dolt_published_storage_and_identifier_evidence() -> None:
     database_url = os.environ.get("OPENSTATSPEC_DOLT_URL")
     if not database_url: