diff --git a/src/dbt_core_interface/dbt_templater/templater.py b/src/dbt_core_interface/dbt_templater/templater.py
index 3ad7557..df8230f 100644
--- a/src/dbt_core_interface/dbt_templater/templater.py
+++ b/src/dbt_core_interface/dbt_templater/templater.py
@@ -45,6 +45,14 @@ def dbt_version(self) -> str:
         """Gets the dbt version."""
         return self._dbt_version.to_version_string()
 
+    def _apply_dbt_builtins(self, config: FluffConfig | None) -> bool:
+        """DbtTemplater uses actual dbt compilation, not FunctionWrapper placeholders.
+
+        Returning False prevents SQLFluff from adding FunctionWrapper objects
+        (which lack .get() method) to the Jinja environment during slice_file analysis.
+        """
+        return False
+
     @large_file_check
     def process(  # pyright: ignore[reportIncompatibleMethodOverride]
         self,
diff --git a/src/dbt_core_interface/project.py b/src/dbt_core_interface/project.py
index 66c338d..cd39925 100644
--- a/src/dbt_core_interface/project.py
+++ b/src/dbt_core_interface/project.py
@@ -9,6 +9,7 @@
 import functools
 import gc
 import json
+import re
 import logging
 import os
 import shlex
@@ -76,7 +77,7 @@ def _set_invocation_context() -> None:
 _set_invocation_context()
 
 logger = logging.getLogger(__name__)
-logger.setLevel(logging.DEBUG)
+logger.setLevel(getattr(logging, os.getenv("LOG_LEVEL", "INFO").upper(), logging.INFO))
 logger.addHandler(rich.logging.RichHandler())
 
 add_logger_to_manager(
@@ -674,6 +675,9 @@ def _create_temp_node(
         self, sql: str, node_id: str | None = None
     ) -> tuple[ManifestNode, t.Callable[[], None]]:
         """Create a temporary node for SQL execution/compilation."""
+        # Remove only the opening and closing snapshot tags, keep the body
+        sql = re.sub(r'{%\s*snapshot\b.*?%}', '', sql, flags=re.DOTALL)
+        sql = re.sub(r'{%\s*endsnapshot\s*%}', '', sql, flags=re.DOTALL)
         node_id = node_id or f"temp_node_{uuid.uuid4().hex[:8]}"
         sql_node = self.sql_parser.parse_remote(sql, node_id)
         process_node(self.runtime_config, self.manifest, sql_node)
diff --git a/src/dbt_core_interface/server.py b/src/dbt_core_interface/server.py
index 1edcd35..4a44cec 100644
--- a/src/dbt_core_interface/server.py
+++ b/src/dbt_core_interface/server.py
@@ -9,6 +9,7 @@
 import json
 import logging
 import os
+import re
 import time
 import typing as t
 import uuid
@@ -284,17 +285,31 @@ def run_sql(
         comp_res = runner.compile_sql(raw_sql)
     try:
         model_context = runner.generate_runtime_model_context(comp_res.node)
-        query = runner.adapter.execute_macro(
-            macro_name="get_show_sql",
-            macro_resolver=runner.manifest,
-            context_override=model_context,
-            kwargs={
-                "compiled_code": model_context["compiled_code"],
-                "sql_header": model_context["config"].get("sql_header"),
-                "limit": limit,
-            },
-        )
+        original_code = model_context["compiled_code"]
+        sql_header = model_context["config"].get("sql_header") or ""
+
+        # Check if query already has a LIMIT clause at the end - if so, execute as-is
+        query: str
+        has_limit = bool(re.search(r"\slimit\s+\d+(\s+offset\s+\d+)?\s*;?\s*$", original_code, re.IGNORECASE))
+
+        if has_limit:
+            query = f"{sql_header}\n{original_code}" if sql_header else original_code
+
+        else:
+            # Wrap in subquery to safely apply limit
+            compiled_code = f"select * from ({original_code}) as __server_query"
+            query = runner.adapter.execute_macro(
+                macro_name="get_show_sql",
+                macro_resolver=runner.manifest,
+                context_override=model_context,
+                kwargs={
+                    "compiled_code": compiled_code,
+                    "sql_header": sql_header,
+                    "limit": limit,
+                },
+            )
         exec_res = runner.execute_sql(t.cast(str, query), compile=False)  # pyright: ignore[reportInvalidCast]
+
     except Exception as e:
         response.status_code = 500
         return ServerErrorContainer(
@@ -469,9 +484,14 @@ def parse_project(
 
 
 @app.get("/health")  # legacy extension support
+def health() -> dict[str, t.Any]:
+    """Health check for vscode-sqlfluff extension. Always returns ready if server is running."""
+    return {"result": {"status": "ready"}}
+
+
 @app.get("/api/v1/status")
 def status(runner: DbtProject = Depends(_get_runner)) -> dict[str, t.Any]:
-    """Health check endpoint to verify server status."""
+    """Status endpoint with project details."""
     return {
         "result": {
             "status": "ready",