diff --git a/pyproject.toml b/pyproject.toml
index 7a771db..e89f5a4 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "model-ledger"
-version = "0.7.9"
+version = "0.7.10"
 description = "Developer-first model inventory and governance framework for SR 11-7, EU AI Act, and NIST AI RMF compliance"
 readme = "README.md"
 requires-python = ">=3.10"
diff --git a/src/model_ledger/backends/snowflake.py b/src/model_ledger/backends/snowflake.py
index 662ff01..1d9d479 100644
--- a/src/model_ledger/backends/snowflake.py
+++ b/src/model_ledger/backends/snowflake.py
@@ -429,14 +429,17 @@ def _flush_snapshots_sql(self) -> None:
             unions = " UNION ALL ".join(
                 f"SELECT {_esc(s.snapshot_hash)}, {_esc(s.model_hash)}, "
                 f"{_esc(s.parent_hash)}, {_esc(s.timestamp.isoformat())}, "
-                f"{_esc(s.actor)}, {_esc(s.event_type)}, {_esc(s.source)}"
+                f"{_esc(s.actor)}, {_esc(s.event_type)}, {_esc(s.source)}, "
+                f"{_esc(json.dumps(s.payload, default=str)) if s.payload else 'NULL'}, "
+                f"{_esc(json.dumps(s.tags, default=str)) if s.tags else 'NULL'}"
                 for s in batch
             )
             self._exec_no_result(
                 f"""
                 INSERT INTO {self._schema}.SNAPSHOTS
-                (SNAPSHOT_HASH, MODEL_HASH, PARENT_HASH, TIMESTAMP, ACTOR, EVENT_TYPE, SOURCE)
-                SELECT * FROM ({unions}) s
+                (SNAPSHOT_HASH, MODEL_HASH, PARENT_HASH, TIMESTAMP, ACTOR, EVENT_TYPE, SOURCE, PAYLOAD, TAGS)
+                SELECT s.$1, s.$2, s.$3, s.$4, s.$5, s.$6, s.$7, PARSE_JSON(s.$8), PARSE_JSON(s.$9)
+                FROM ({unions}) s
                 WHERE NOT EXISTS (SELECT 1 FROM {self._schema}.SNAPSHOTS t WHERE t.SNAPSHOT_HASH = s.$1)""",
             )
 
diff --git a/tests/test_backends/test_snowflake_ledger.py b/tests/test_backends/test_snowflake_ledger.py
index 5784dc1..d72346c 100644
--- a/tests/test_backends/test_snowflake_ledger.py
+++ b/tests/test_backends/test_snowflake_ledger.py
@@ -326,6 +326,79 @@ def sql(self, query: str, params: Any = None) -> MockCollectResult:
     )
 
 
+def test_snapshot_sql_fallback_preserves_payload_and_tags():
+    """The DDL-free INSERT fallback must carry PAYLOAD and TAGS. A
+    least-privilege role (INSERT/UPDATE only, no CREATE TEMPORARY TABLE)
+    always lands on this path; omitting the columns silently persists
+    snapshots with NULL payloads — losing the event data itself.
+    """
+    from model_ledger.backends.snowflake import SnowflakeLedgerBackend
+
+    inserts: list[str] = []
+
+    class RecordingSession:
+        """Lacks a pandas-capable connection, forcing the SQL fallback."""
+
+        def sql(self, query: str, params: Any = None) -> MockCollectResult:
+            upper = query.upper()
+            if "INSERT INTO" in upper and ".SNAPSHOTS" in upper:
+                inserts.append(query)
+            return MockCollectResult([])
+
+    backend = SnowflakeLedgerBackend(schema="TEST_SCHEMA", connection=RecordingSession())
+    backend.append_snapshot(
+        Snapshot(
+            snapshot_hash="snap-payload-1",
+            model_hash="m1",
+            timestamp=datetime(2025, 1, 1, tzinfo=timezone.utc),
+            actor="scanner",
+            event_type="metadata_inferred",
+            source="alerting",
+            payload={"queue_slug": "case_review_p1", "threshold": 0.85},
+            tags={"cycle": "2026"},
+        )
+    )
+    backend.flush()
+
+    assert len(inserts) == 1, f"expected one SNAPSHOTS INSERT, saw {len(inserts)}"
+    sql = inserts[0]
+    assert "PAYLOAD" in sql and "TAGS" in sql, "fallback INSERT omits PAYLOAD/TAGS columns"
+    assert "PARSE_JSON" in sql, "payload/tags must be parsed into VARIANT"
+    assert "queue_slug" in sql, "payload content missing from INSERT source"
+    assert "cycle" in sql, "tags content missing from INSERT source"
+
+
+def test_snapshot_sql_fallback_null_payload_stays_null():
+    """Snapshots without payload/tags must insert SQL NULL, not the string 'null'."""
+    from model_ledger.backends.snowflake import SnowflakeLedgerBackend
+
+    inserts: list[str] = []
+
+    class RecordingSession:
+        def sql(self, query: str, params: Any = None) -> MockCollectResult:
+            upper = query.upper()
+            if "INSERT INTO" in upper and ".SNAPSHOTS" in upper:
+                inserts.append(query)
+            return MockCollectResult([])
+
+    backend = SnowflakeLedgerBackend(schema="TEST_SCHEMA", connection=RecordingSession())
+    backend.append_snapshot(
+        Snapshot(
+            snapshot_hash="snap-nopayload-1",
+            model_hash="m1",
+            timestamp=datetime(2025, 1, 1, tzinfo=timezone.utc),
+            actor="scanner",
+            event_type="registered",
+            source="alerting",
+        )
+    )
+    backend.flush()
+
+    assert len(inserts) == 1
+    row_source = inserts[0].split("FROM")[0]
+    assert "'null'" not in row_source.lower(), "empty payload serialized as JSON 'null' string"
+
+
 class TestStatusPropagationSQL:
     """Connector-discovered status must land in the MODELS table via the MERGE.