Quoted and Capitalised table names handling

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "motley-slayer"
-version = "0.7.4"
+version = "0.7.5"
 description = "A lightweight, agent-first semantic layer for AI agents"
 requires-python = ">=3.11"
 license = "MIT"

slayer/engine/schema_drift.py

@@ -1751,17 +1751,37 @@ async def _live_columns_for_sql_model(
 # ===========================================================================
 
 
+def _strip_ident_quotes(ident: str) -> str:
+    """Strip surrounding double-quotes from an SQL identifier and unescape
+    ``""`` → ``"``. Bare identifiers pass through unchanged.
+    """
+    ident = ident.strip()
+    if len(ident) >= 2 and ident[0] == '"' == ident[-1]:
+        return ident[1:-1].replace('""', '"')
+    return ident
+
+
 def _resolve_live_table(
     *, sql_table: str, live_tables: Dict[str, LiveTable]
 ) -> Optional[LiveTable]:
     """Look up a model's ``sql_table`` in the live introspection map,
     falling back to the bare name when the persisted value is schema-
-    qualified (``schema.table``).
+    qualified (``schema.table``) and unquoting double-quoted identifiers
+    (e.g. ``prod."Company"`` for case-sensitive Postgres tables).
     """
-    live = live_tables.get(sql_table)
-    if live is None and "." in sql_table:
-        live = live_tables.get(sql_table.split(".", 1)[1])
-    return live
+    candidates = [sql_table]
+    if "." in sql_table:
+        candidates.append(sql_table.split(".", 1)[1])
+    # Materialise the snapshot before extending — a bare generator
+    # ``(_strip_ident_quotes(c) for c in candidates)`` would iterate the
+    # list lazily WHILE ``extend`` appends to it, so every appended item
+    # gets re-fed into the iterator and the loop never terminates.
+    candidates.extend([_strip_ident_quotes(c) for c in candidates])
+    for name in candidates:
+        live = live_tables.get(name)
+        if live is not None:
+            return live
+    return None
 
 
 def _is_validate_models_base_column(col: Column) -> bool:

tests/test_validate_models.py

@@ -38,6 +38,8 @@
     LiveTable,
     RemoveSpec,
     WholeModelDelete,
+    _resolve_live_table,
+    _strip_ident_quotes,
     compute_datasource_drops,
     data_type_bucket,
     diff_sql_model,
@@ -1215,3 +1217,64 @@ async def test_probe_drift_cascades_to_derived_column(
         # Both the base column and the derived column are dropped.
         assert "tempstabidx" in entry.remove.columns
         assert "doubled" in entry.remove.columns
+
+
+class TestStripIdentQuotes:
+    def test_bare_identifier_unchanged(self) -> None:
+        assert _strip_ident_quotes("Company") == "Company"
+
+    def test_double_quoted(self) -> None:
+        assert _strip_ident_quotes('"Company"') == "Company"
+
+    def test_escaped_inner_quote(self) -> None:
+        assert _strip_ident_quotes('"Comp""any"') == 'Comp"any'
+
+    def test_unbalanced_quotes_left_alone(self) -> None:
+        assert _strip_ident_quotes('"Company') == '"Company'
+
+    def test_surrounding_whitespace_stripped(self) -> None:
+        assert _strip_ident_quotes('  "Company"  ') == "Company"
+
+
+class TestResolveLiveTable:
+    """Live-table lookup must handle schema-qualified and quoted identifiers."""
+
+    @staticmethod
+    def _live(name: str) -> LiveTable:
+        return LiveTable(columns={"id": DataType.DOUBLE}, pk_columns={"id"})
+
+    def test_bare_table_matches(self) -> None:
+        live = self._live("orders")
+        assert _resolve_live_table(
+            sql_table="orders", live_tables={"orders": live}
+        ) is live
+
+    def test_schema_qualified_falls_back_to_bare(self) -> None:
+        live = self._live("orders")
+        assert _resolve_live_table(
+            sql_table="public.orders", live_tables={"orders": live}
+        ) is live
+
+    def test_quoted_table_part_strips_quotes(self) -> None:
+        live = self._live("Company")
+        # The bug's repro: prod."Company" with introspection keyed by bare name.
+        assert _resolve_live_table(
+            sql_table='prod."Company"', live_tables={"Company": live}
+        ) is live
+
+    def test_both_parts_quoted(self) -> None:
+        live = self._live("Company")
+        assert _resolve_live_table(
+            sql_table='"prod"."Company"', live_tables={"Company": live}
+        ) is live
+
+    def test_escaped_inner_quote_in_table(self) -> None:
+        live = self._live('Comp"any')
+        assert _resolve_live_table(
+            sql_table='prod."Comp""any"', live_tables={'Comp"any': live}
+        ) is live
+
+    def test_missing_returns_none(self) -> None:
+        assert _resolve_live_table(
+            sql_table='prod."Missing"', live_tables={"orders": self._live("orders")}
+        ) is None