fix: pandas bulk path falls back to SQL on any failure (poisoned-buffer 500 loop)

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "model-ledger"
-version = "0.7.10"
+version = "0.7.11"
 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"

src/model_ledger/backends/snowflake.py

@@ -6,14 +6,18 @@
 
 from __future__ import annotations
 
+import contextlib
 import json
+import logging
 import threading
 from collections.abc import Callable
 from datetime import datetime, timezone
 from typing import Any
 
 from model_ledger.core.ledger_models import ModelRef, Snapshot, Tag
 
+logger = logging.getLogger(__name__)
+
 BATCH_SIZE = 50
 
 # Snowflake error code raised when a session's auth token has idle-expired
@@ -317,9 +321,23 @@ def _flush_models_pandas(self) -> bool:
             )
             self._exec_no_result(f"DROP TABLE IF EXISTS {staging}")
         except Exception as e:
-            if _is_privilege_error(e):
-                return False
-            raise
+            # The pandas path is an optimization, not a correctness
+            # requirement: privilege denials (no CREATE TEMPORARY TABLE) and
+            # transport limitations (sessions whose backing protocol cannot
+            # run PUT file transfers — write_pandas dies inside
+            # file_transfer_agent, e.g. KeyError('command')) both land here.
+            # Any failure falls back to the DDL-free SQL path, which is
+            # idempotent and carries all columns. Never leave the buffer
+            # poisoned: a raised exception here would re-fire on every
+            # subsequent flush-before-read and 500 the whole API.
+            logger.warning(
+                "pandas bulk path failed (%s: %s); falling back to SQL flush",
+                type(e).__name__,
+                e,
+            )
+            with contextlib.suppress(Exception):
+                self._exec_no_result(f"DROP TABLE IF EXISTS {staging}")
+            return False
         return True
 
     def _flush_models_sql(self) -> None:
@@ -415,9 +433,23 @@ def _flush_snapshots_pandas(self) -> bool:
             )
             self._exec_no_result(f"DROP TABLE IF EXISTS {staging}")
         except Exception as e:
-            if _is_privilege_error(e):
-                return False
-            raise
+            # The pandas path is an optimization, not a correctness
+            # requirement: privilege denials (no CREATE TEMPORARY TABLE) and
+            # transport limitations (sessions whose backing protocol cannot
+            # run PUT file transfers — write_pandas dies inside
+            # file_transfer_agent, e.g. KeyError('command')) both land here.
+            # Any failure falls back to the DDL-free SQL path, which is
+            # idempotent and carries all columns. Never leave the buffer
+            # poisoned: a raised exception here would re-fire on every
+            # subsequent flush-before-read and 500 the whole API.
+            logger.warning(
+                "pandas bulk path failed (%s: %s); falling back to SQL flush",
+                type(e).__name__,
+                e,
+            )
+            with contextlib.suppress(Exception):
+                self._exec_no_result(f"DROP TABLE IF EXISTS {staging}")
+            return False
         return True
 
     def _flush_snapshots_sql(self) -> None:

tests/test_backends/test_snowflake_ledger.py

@@ -1066,3 +1066,59 @@ def test_requires_connection_or_factory(self):
 
         with pytest.raises(ValueError, match="connection"):
             SnowflakeLedgerBackend(schema="TEST_SCHEMA")
+
+
+def test_pandas_path_failure_falls_back_to_sql(monkeypatch):
+    """A pandas bulk-path failure that is NOT a privilege error (e.g. the
+    backing session's protocol cannot run PUT file transfers — write_pandas
+    raises KeyError('command') from file_transfer_agent) must fall back to
+    the SQL path instead of raising. A raised exception leaves the buffer
+    poisoned: every subsequent flush-before-read re-fires it and the whole
+    API 500s (observed in production 2026-07-10).
+    """
+    import sys
+    import types
+
+    from model_ledger.backends.snowflake import SnowflakeLedgerBackend
+
+    statements: list[str] = []
+
+    class FakeConn:
+        _session_parameters = {"QUERY_TAG": "test"}
+
+    class PandasCapableSession:
+        """Looks pandas-capable (has _connection w/ _session_parameters)."""
+
+        _connection = FakeConn()
+
+        def sql(self, query: str, params: Any = None) -> MockCollectResult:
+            statements.append(query)
+            return MockCollectResult([[0]])
+
+    def exploding_write_pandas(*args, **kwargs):
+        raise KeyError("command")
+
+    fake_tools = types.ModuleType("snowflake.connector.pandas_tools")
+    fake_tools.write_pandas = exploding_write_pandas
+    monkeypatch.setitem(sys.modules, "snowflake.connector.pandas_tools", fake_tools)
+
+    backend = SnowflakeLedgerBackend(schema="TEST_SCHEMA", connection=PandasCapableSession())
+    backend.save_model(
+        ModelRef(
+            name="fraud_scorer",
+            owner="risk-team",
+            model_type="scoring_model",
+            tier="unclassified",
+            purpose="",
+        )
+    )
+    backend.flush()  # must NOT raise
+
+    merges = [s for s in statements if "MERGE INTO" in s.upper() and ".MODELS" in s.upper()]
+    assert merges, "SQL fallback MERGE never ran after pandas-path failure"
+    assert not backend._model_buffer, "buffer not cleared — would re-fire on every flush"
+
+    # Second flush must be a no-op, not a re-explosion.
+    before = len(statements)
+    backend.flush()
+    assert len(statements) == before