Repository: TobikoData/sqlmesh
Pull request #4510 (2025-05-22 18:05:43)

Feat: expose model fqn in the macro evaluator
There's no easy way to access the unresolved name of a model within a macro that it calls– `this_model` is usually the resolved version, corresponding to either a physical table, or a view if we're in the promoting stage.

This PR adds a new property in the macro evaluator that can be used to access the model's fully-qualified name. One example where the unresolved name is helpful can be seen in the added test.

georgesittas: Yep, I'll update it.

Patch:
diff --git a/Makefile b/Makefile
index 7c2aa11d..c99936ef 100644
--- a/Makefile
+++ b/Makefile
@@ -170,7 +170,7 @@ clickhouse-cloud-test: guard-CLICKHOUSE_CLOUD_HOST guard-CLICKHOUSE_CLOUD_USERNA
 	pytest -n 1 -m "clickhouse_cloud" --retries 3 --junitxml=test-results/junit-clickhouse-cloud.xml
 
 athena-test: guard-AWS_ACCESS_KEY_ID guard-AWS_SECRET_ACCESS_KEY guard-ATHENA_S3_WAREHOUSE_LOCATION engine-athena-install
-	pytest -n auto -m "athena" --retries 3 --retry-delay 10 --junitxml=test-results/junit-athena.xml
+	pytest -n auto -m "athena" --retries 3 --junitxml=test-results/junit-athena.xml
 
 vscode_settings:
 	mkdir -p .vscode
diff --git a/examples/custom_materializations/pyproject.toml b/examples/custom_materializations/pyproject.toml
index b680813c..70eb2d7f 100644
--- a/examples/custom_materializations/pyproject.toml
+++ b/examples/custom_materializations/pyproject.toml
@@ -1,3 +1,16 @@
 [build-system]
 requires = ["setuptools >= 61.0", "setuptools_scm"]
 build-backend = "setuptools.build_meta"
+
+[project]
+name = "custom_materializations"
+requires-python = ">=3.9"
+version = "0.1.0"
+dependencies = ["sqlmesh"]
+
+[project.entry-points."sqlmesh.materializations"]
+custom_full_materialization = "custom_materializations.full:CustomFullMaterialization"
+custom_full_with_custom_kind = "custom_materializations.custom_kind:CustomFullWithCustomKindMaterialization"
+
+[tool.setuptools.packages.find]
+include = ["custom_materializations"]
diff --git a/examples/custom_materializations/setup.py b/examples/custom_materializations/setup.py
deleted file mode 100644
index b90200be..00000000
--- a/examples/custom_materializations/setup.py
+++ /dev/null
@@ -1,15 +0,0 @@
-from setuptools import find_packages, setup
-
-setup(
-    name="custom_materializations",
-    packages=find_packages(include=["custom_materializations"]),
-    entry_points={
-        "sqlmesh.materializations": [
-            "custom_full_materialization = custom_materializations.full:CustomFullMaterialization",
-            "custom_full_with_custom_kind = custom_materializations.custom_kind:CustomFullWithCustomKindMaterialization",
-        ],
-    },
-    install_requires=[
-        "sqlmesh",
-    ],
-)
diff --git a/pyproject.toml b/pyproject.toml
index c84f6b2a..e2538056 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -215,6 +215,53 @@ module = [
 ]
 ignore_missing_imports = true
 
+[tool.pytest.ini_options]
+markers = [
+    # Test Type Markers
+    # Tests are ordered from fastest to slowest
+    "fast: fast tests (automatically applied if no type markers)",
+    "slow: slow tests that typically involve interacting with a local DB (like DuckDB)",
+    "docker: test that involves interacting with a Docker container",
+    "remote: test that involves interacting with a remote DB",
+    "cicdonly: test that only runs on CI/CD",
+    "isolated: tests that need to run sequentially usually because they use fork",
+
+    # Test Domain Markers
+    # default: core functionality
+    "cli: test for CLI",
+    "dbt: test for dbt adapter",
+    "github: test for Github CI/CD bot",
+    "jupyter: tests for Jupyter integration",
+    "web: tests for web UI",
+
+    # Engine Adapters
+    "engine: test all engine adapters",
+    "athena: test for Athena",
+    "bigquery: test for BigQuery",
+    "clickhouse: test for Clickhouse (standalone mode / cluster mode)",
+    "clickhouse_cloud: test for Clickhouse (cloud mode)",
+    "databricks: test for Databricks",
+    "duckdb: test for DuckDB",
+    "motherduck: test for MotherDuck",
+    "mssql: test for MSSQL",
+    "mysql: test for MySQL",
+    "postgres: test for Postgres",
+    "redshift: test for Redshift",
+    "snowflake: test for Snowflake",
+    "spark: test for Spark",
+    "trino: test for Trino (all connectors)",
+    "risingwave: test for Risingwave"
+]
+addopts = "-n 0 --dist=loadgroup"
+asyncio_default_fixture_loop_scope = "session"
+log_cli = false # Set this to true to enable logging during tests
+log_cli_format = "%(asctime)s.%(msecs)03d %(filename)s:%(lineno)d %(levelname)s %(message)s"
+log_cli_level = "INFO"
+filterwarnings = [
+    "ignore:The localize method is no longer necessary, as this time zone supports the fold attribute"
+]
+retry_delay = 10
+
 [tool.ruff.lint]
 select = [
     "F401",
diff --git a/sqlmesh/__init__.py b/sqlmesh/__init__.py
index e48ffbef..63f109d4 100644
--- a/sqlmesh/__init__.py
+++ b/sqlmesh/__init__.py
@@ -166,11 +166,24 @@ class CustomFormatter(logging.Formatter):
         return formatter.format(record)
 
 
+def remove_excess_logs(
+    log_file_dir: t.Optional[t.Union[str, Path]] = None,
+    log_limit: int = c.DEFAULT_LOG_LIMIT,
+) -> None:
+    if log_limit <= 0:
+        return
+
+    log_file_dir = log_file_dir or c.DEFAULT_LOG_FILE_DIR
+    log_path_prefix = Path(log_file_dir) / LOG_FILENAME_PREFIX
+
+    for path in list(sorted(glob.glob(f"{log_path_prefix}*.log"), reverse=True))[log_limit:]:
+        os.remove(path)
+
+
 def configure_logging(
     force_debug: bool = False,
     write_to_stdout: bool = False,
     write_to_file: bool = True,
-    log_limit: int = c.DEFAULT_LOG_LIMIT,
     log_file_dir: t.Optional[t.Union[str, Path]] = None,
     ignore_warnings: bool = False,
 ) -> None:
@@ -192,20 +205,18 @@ def configure_logging(
 
     log_file_dir = log_file_dir or c.DEFAULT_LOG_FILE_DIR
     log_path_prefix = Path(log_file_dir) / LOG_FILENAME_PREFIX
+
     if write_to_file:
         os.makedirs(str(log_file_dir), exist_ok=True)
         filename = f"{log_path_prefix}{datetime.now().strftime('%Y_%m_%d_%H_%M_%S')}.log"
         file_handler = logging.FileHandler(filename, mode="w", encoding="utf-8")
+
         # the log files should always log at least info so that users will always have
         # minimal info for debugging even if they specify "ignore_warnings"
         file_handler.setLevel(level)
         file_handler.setFormatter(logging.Formatter(LOG_FORMAT))
         logger.addHandler(file_handler)
 
-    if log_limit > 0:
-        for path in list(sorted(glob.glob(f"{log_path_prefix}*.log"), reverse=True))[log_limit:]:
-            os.remove(path)
-
     if debug:
         import faulthandler
 
diff --git a/sqlmesh/cli/main.py b/sqlmesh/cli/main.py
index 5b5fcf15..2ef9e254 100644
--- a/sqlmesh/cli/main.py
+++ b/sqlmesh/cli/main.py
@@ -7,7 +7,7 @@ import typing as t
 
 import click
 
-from sqlmesh import configure_logging
+from sqlmesh import configure_logging, remove_excess_logs
 from sqlmesh.cli import error_handler
 from sqlmesh.cli import options as opt
 from sqlmesh.cli.example_project import ProjectTemplate, init_example_project
@@ -100,17 +100,19 @@ def cli(
         if ctx.invoked_subcommand in SKIP_LOAD_COMMANDS:
             load = False
 
-    configs = load_configs(config, Context.CONFIG_TYPE, paths)
-    log_limit = list(configs.values())[0].log_limit
     configure_logging(
         debug,
         log_to_stdout,
-        log_limit=log_limit,
         log_file_dir=log_file_dir,
         ignore_warnings=ignore_warnings,
     )
     configure_console(ignore_warnings=ignore_warnings)
 
+    configs = load_configs(config, Context.CONFIG_TYPE, paths)
+    log_limit = list(configs.values())[0].log_limit
+
+    remove_excess_logs(log_file_dir, log_limit)
+
     try:
         context = Context(
             paths=paths,
diff --git a/sqlmesh/core/config/connection.py b/sqlmesh/core/config/connection.py
index 1e1afc05..18597945 100644
--- a/sqlmesh/core/config/connection.py
+++ b/sqlmesh/core/config/connection.py
@@ -13,6 +13,7 @@ from functools import partial
 
 import pydantic
 from pydantic import Field
+from pydantic_core import from_json
 from packaging import version
 from sqlglot import exp
 from sqlglot.helper import subclasses
@@ -26,13 +27,14 @@ from sqlmesh.core.config.common import (
 )
 from sqlmesh.core.engine_adapter.shared import CatalogSupport
 from sqlmesh.core.engine_adapter import EngineAdapter
-from sqlmesh.utils import str_to_bool
+from sqlmesh.utils import debug_mode_enabled, str_to_bool
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.pydantic import (
     ValidationInfo,
     field_validator,
     model_validator,
     validation_error_message,
+    get_concrete_types_from_typehint,
 )
 from sqlmesh.utils.aws import validate_s3_uri
 
@@ -67,8 +69,16 @@ def _get_engine_import_validator(
         try:
             importlib.import_module(import_name)
         except ImportError:
+            if debug_mode_enabled():
+                raise
+
+            logger.exception("Failed to import the engine library")
+
             raise ConfigError(
-                f"Failed to import the '{engine_type}' engine library. Please run `pip install \"sqlmesh[{extra_name}]\"`."
+                f"Failed to import the '{engine_type}' engine library. This may be due to a missing "
+                "or incompatible installation. Please ensure the required dependency is installed by "
+                f'running: `pip install "sqlmesh[{extra_name}]"`. For more details, check the logs '
+                "in the 'logs/' folder, or rerun the command with the '--debug' flag."
             )
 
         return data
@@ -169,6 +179,42 @@ class ConnectionConfig(abc.ABC, BaseConfig):
             return self.db
         return None
 
+    @model_validator(mode="before")
+    @classmethod
+    def _expand_json_strings_to_concrete_types(cls, data: t.Any) -> t.Any:
+        """
+        There are situations where a connection config class has a field that is some kind of complex type
+        (eg a list of strings or a dict) but the value is being supplied from a source such as an environment variable
+
+        When this happens, the value is supplied as a string rather than a Python object. We need some way
+        of turning this string into the corresponding Python list or dict.
+
+        Rather than doing this piecemeal on every config subclass, this provides a generic implementatation
+        to identify fields that may be be supplied as JSON strings and handle them transparently
+        """
+        if data and isinstance(data, dict):
+            for maybe_json_field_name in cls._get_list_and_dict_field_names():
+                if (value := data.get(maybe_json_field_name)) and isinstance(value, str):
+                    # crude JSON check as we dont want to try and parse every string we get
+                    value = value.strip()
+                    if value.startswith("{") or value.startswith("["):
+                        data[maybe_json_field_name] = from_json(value)
+
+        return data
+
+    @classmethod
+    def _get_list_and_dict_field_names(cls) -> t.Set[str]:
+        field_names = set()
+        for name, field in cls.model_fields.items():
+            if field.annotation:
+                field_types = get_concrete_types_from_typehint(field.annotation)
+
+                # check if the field type is something that could concievably be supplied as a json string
+                if any(ft is t for t in (list, tuple, set, dict) for ft in field_types):
+                    field_names.add(name)
+
+        return field_names
+
 
 class DuckDBAttachOptions(BaseConfig):
     type: str
diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
index f53e9843..b5c9f7b6 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -35,7 +35,6 @@ from __future__ import annotations
 
 import abc
 import collections
-from itertools import chain
 import logging
 import sys
 import time
@@ -44,6 +43,7 @@ import typing as t
 import unittest.result
 from functools import cached_property
 from io import StringIO
+from itertools import chain
 from pathlib import Path
 from shutil import rmtree
 from types import MappingProxyType
diff --git a/sqlmesh/core/macros.py b/sqlmesh/core/macros.py
index 5326804a..db530481 100644
--- a/sqlmesh/core/macros.py
+++ b/sqlmesh/core/macros.py
@@ -173,6 +173,7 @@ class MacroEvaluator:
         default_catalog: t.Optional[str] = None,
         path: Path = Path(),
         environment_naming_info: t.Optional[EnvironmentNamingInfo] = None,
+        model_fqn: t.Optional[str] = None,
     ):
         self.dialect = dialect
         self.generator = MacroDialect().generator()
@@ -198,6 +199,7 @@ class MacroEvaluator:
         self._snapshots = snapshots if snapshots is not None else {}
         self._path = path
         self._environment_naming_info = environment_naming_info
+        self._model_fqn = model_fqn
 
         prepare_env(self.python_env, self.env)
         for k, v in self.python_env.items():
@@ -476,6 +478,12 @@ class MacroEvaluator:
             raise SQLMeshError("Model name is not available in the macro evaluator.")
         return this_model.sql(dialect=self.dialect, identify=True, comments=False)
 
+    @property
+    def this_model_fqn(self) -> str:
+        if self._model_fqn is None:
+            raise SQLMeshError("Model name is not available in the macro evaluator.")
+        return self._model_fqn
+
     @property
     def engine_adapter(self) -> EngineAdapter:
         engine_adapter = self.locals.get("engine_adapter")
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index c19cf5d7..8a282636 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -2488,22 +2488,27 @@ def _create_model(
 
     model.audit_definitions.update(audit_definitions)
 
-    from sqlmesh.core.audit.builtin import BUILT_IN_AUDITS
+    # Any macro referenced in audits or signals needs to be treated as metadata-only
+    statements.extend((audit.query, True) for audit in audit_definitions.values())
 
     # Ensure that all audits referenced in the model are defined
+    from sqlmesh.core.audit.builtin import BUILT_IN_AUDITS
+
     available_audits = BUILT_IN_AUDITS.keys() | model.audit_definitions.keys()
-    for referenced_audit, *_ in model.audits:
+    for referenced_audit, audit_args in model.audits:
         if referenced_audit not in available_audits:
             raise_config_error(f"Audit '{referenced_audit}' is undefined", location=path)
 
-    # Any macro referenced in audits or signals needs to be treated as metadata-only
-    statements.extend((audit.query, True) for audit in audit_definitions.values())
-    for _, audit_args in model.audits:
         statements.extend(
             (audit_arg_expression, True) for audit_arg_expression in audit_args.values()
         )
 
-    for _, kwargs in model.signals:
+    signal_definitions = signal_definitions or UniqueKeyDict("signals")
+
+    for referenced_signal, kwargs in model.signals:
+        if referenced_signal and referenced_signal not in signal_definitions:
+            raise_config_error(f"Signal '{referenced_signal}' is undefined", location=path)
+
         statements.extend((signal_kwarg, True) for signal_kwarg in kwargs.values())
 
     python_env = make_python_env(
@@ -2523,7 +2528,7 @@ def _create_model(
     env: t.Dict[str, t.Tuple[t.Any, t.Optional[bool]]] = {}
 
     for signal_name, _ in model.signals:
-        if signal_definitions and signal_name in signal_definitions:
+        if signal_name and signal_name in signal_definitions:
             func = signal_definitions[signal_name].func
             setattr(func, c.SQLMESH_METADATA, True)
             build_env(func, env=env, name=signal_name, path=module_path)
diff --git a/sqlmesh/core/plan/builder.py b/sqlmesh/core/plan/builder.py
index eed05e04..0e8bc481 100644
--- a/sqlmesh/core/plan/builder.py
+++ b/sqlmesh/core/plan/builder.py
@@ -344,6 +344,15 @@ class PlanBuilder:
         for s_id in dag:
             snapshot = self._context_diff.snapshots[s_id]
 
+            # Since we are traversing the graph in topological order and the largest interval range is pushed down
+            # the graph we just have to check our immediate parents in the graph and not the whole upstream graph.
+            restating_parents = [
+                self._context_diff.snapshots[s] for s in snapshot.parents if s in restatements
+            ]
+
+            if not restating_parents and snapshot.name not in restate_models:
+                continue
+
             if not forward_only_preview_needed:
                 if self._is_dev and not snapshot.is_paused:
                     self._console.log_warning(
@@ -362,15 +371,6 @@ class PlanBuilder:
                     logger.info("Skipping restatement for model '%s'", snapshot.name)
                     continue
 
-            # Since we are traversing the graph in topological order and the largest interval range is pushed down
-            # the graph we just have to check our immediate parents in the graph and not the whole upstream graph.
-            restating_parents = [
-                self._context_diff.snapshots[s] for s in snapshot.parents if s in restatements
-            ]
-
-            if not restating_parents and snapshot.name not in restate_models:
-                continue
-
             possible_intervals = {
                 restatements[p.snapshot_id] for p in restating_parents if p.is_incremental
             }
diff --git a/sqlmesh/core/plan/evaluator.py b/sqlmesh/core/plan/evaluator.py
index 0400e6cb..0d2a25b4 100644
--- a/sqlmesh/core/plan/evaluator.py
+++ b/sqlmesh/core/plan/evaluator.py
@@ -116,6 +116,7 @@ class BuiltInPlanEvaluator(PlanEvaluator):
                 after_promote_snapshots = all_names - before_promote_snapshots
                 deployability_index_for_evaluation = DeployabilityIndex.all_deployable()
 
+            # Step 1: Run before_all environment statements before doing anything else
             execute_environment_statements(
                 adapter=self.snapshot_evaluator.adapter,
                 environment_statements=plan.environment_statements or [],
@@ -128,15 +129,24 @@ class BuiltInPlanEvaluator(PlanEvaluator):
                 execution_time=plan.execution_time,
             )
 
-            self._run_audits_for_metadata_snapshots(plan, new_snapshots)
-
+            # Step 2:Store new snapshot records in the state and create physical tables for them
             push_completion_status = self._push(plan, snapshots, deployability_index_for_creation)
             if push_completion_status.is_nothing_to_do:
                 self.console.log_success(
                     "" if plan.restatements else "\nSKIP: No physical layer updates to perform"
                 )
+
+            # Step 3: Update the intervals for the new forward-only snapshots
             update_intervals_for_new_snapshots(plan.new_snapshots, self.state_sync)
+
+            # Step 4: Run audits without evaluations for snapshots that capture audit metadata changes
+            self._run_audits_for_metadata_snapshots(plan, new_snapshots)
+
+            # Step 5: Remove intervals for snapshots that need to be restated
             self._restate(plan, snapshots_by_name)
+
+            # Step 6: Backfill missing intervals for snapshots that can be backfilled before updating
+            # the schema of production tables
             first_bf_completion_status = self._backfill(
                 plan,
                 snapshots_by_name,
@@ -144,9 +154,15 @@ class BuiltInPlanEvaluator(PlanEvaluator):
                 deployability_index_for_evaluation,
                 circuit_breaker=circuit_breaker,
             )
+
+            # Step 7: Update the target environment record in the state and migrate table schemas for forward-only
+            # snapshots if deploying to production
             promotion_result = self._promote(
                 plan, snapshots, before_promote_snapshots, deployability_index_for_creation
             )
+
+            # Step 8: Backfill missing intervals for snapshots that can be backfilled only after updating
+            # the schema of production tables
             second_bf_completion_status = self._backfill(
                 plan,
                 snapshots_by_name,
@@ -154,15 +170,20 @@ class BuiltInPlanEvaluator(PlanEvaluator):
                 deployability_index_for_evaluation,
                 circuit_breaker=circuit_breaker,
             )
+
             if (
                 first_bf_completion_status.is_nothing_to_do
                 and second_bf_completion_status.is_nothing_to_do
             ):
                 self.console.log_success("SKIP: No model batches to execute")
+
+            # Step 9: Update environment views to point at new physical tables and finalize the environment
+            # record in the state
             self._update_views(
                 plan, snapshots, promotion_result, deployability_index_for_evaluation
             )
 
+            # Step 10: Run after_all environment statements
             execute_environment_statements(
                 adapter=self.snapshot_evaluator.adapter,
                 environment_statements=plan.environment_statements or [],
diff --git a/sqlmesh/core/renderer.py b/sqlmesh/core/renderer.py
index fa07c209..c683fc58 100644
--- a/sqlmesh/core/renderer.py
+++ b/sqlmesh/core/renderer.py
@@ -169,6 +169,7 @@ class BaseExpressionRenderer:
             default_catalog=self._default_catalog,
             path=self._path,
             environment_naming_info=environment_naming_info,
+            model_fqn=self._model_fqn,
         )
 
         start_time, end_time = (
diff --git a/sqlmesh/lsp/main.py b/sqlmesh/lsp/main.py
index 59575cda..c3d17d26 100644
--- a/sqlmesh/lsp/main.py
+++ b/sqlmesh/lsp/main.py
@@ -69,10 +69,7 @@ class SQLMeshLanguageServer:
                                     # Use user-provided instantiator to build the context
                                     created_context = self.context_class(paths=[folder_path])
                                     self.lsp_context = LSPContext(created_context)
-                                    ls.show_message(
-                                        f"Loaded SQLMesh context from {config_path}",
-                                        types.MessageType.Info,
-                                    )
+                                    loaded_sqlmesh_message(ls, folder_path)
                                     return  # Exit after successfully loading any config
                                 except Exception as e:
                                     ls.show_message(
@@ -94,6 +91,9 @@ class SQLMeshLanguageServer:
         @self.server.feature(API_FEATURE)
         def api(ls: LanguageServer, request: ApiRequest) -> t.Dict[str, t.Any]:
             ls.log_trace(f"API request: {request}")
+            if self.lsp_context is None:
+                current_path = Path.cwd()
+                self._ensure_context_in_folder(current_path)
             if self.lsp_context is None:
                 raise RuntimeError("No context found")
 
@@ -291,6 +291,20 @@ class SQLMeshLanguageServer:
             raise RuntimeError("No context found")
         return self.lsp_context
 
+    def _ensure_context_in_folder(self, folder_uri: Path) -> None:
+        if self.lsp_context is not None:
+            return
+        for ext in ("py", "yml", "yaml"):
+            config_path = folder_uri / f"config.{ext}"
+            if config_path.exists():
+                try:
+                    created_context = self.context_class(paths=[folder_uri])
+                    self.lsp_context = LSPContext(created_context)
+                    loaded_sqlmesh_message(self.server, folder_uri)
+                    return
+                except Exception as e:
+                    self.server.show_message(f"Error loading context: {e}", types.MessageType.Error)
+
     def _ensure_context_for_document(
         self,
         document_uri: URI,
@@ -382,6 +396,13 @@ class SQLMeshLanguageServer:
         self.server.start_io()
 
 
+def loaded_sqlmesh_message(ls: LanguageServer, folder: Path) -> None:
+    ls.show_message(
+        f"Loaded SQLMesh context from {folder}",
+        types.MessageType.Info,
+    )
+
+
 def main() -> None:
     # Example instantiator that just uses the same signature as your original `Context` usage.
     sqlmesh_server = SQLMeshLanguageServer(context_class=Context)
diff --git a/sqlmesh/magics.py b/sqlmesh/magics.py
index c79f6bb4..6929cc81 100644
--- a/sqlmesh/magics.py
+++ b/sqlmesh/magics.py
@@ -129,18 +129,23 @@ class SQLMeshMagics(Magics):
     @line_magic
     def context(self, line: str) -> None:
         """Sets the context in the user namespace."""
-        from sqlmesh import configure_logging
+        from sqlmesh import configure_logging, remove_excess_logs
 
         args = parse_argstring(self.context, line)
-        configs = load_configs(args.config, Context.CONFIG_TYPE, args.paths)
-        log_limit = list(configs.values())[0].log_limit
+        log_file_dir = args.log_file_dir
+
         configure_logging(
             args.debug,
-            log_limit=log_limit,
-            log_file_dir=args.log_file_dir,
+            log_file_dir=log_file_dir,
             ignore_warnings=args.ignore_warnings,
         )
         configure_console(ignore_warnings=args.ignore_warnings)
+
+        configs = load_configs(args.config, Context.CONFIG_TYPE, args.paths)
+        log_limit = list(configs.values())[0].log_limit
+
+        remove_excess_logs(log_file_dir, log_limit)
+
         try:
             context = Context(paths=args.paths, config=configs, gateway=args.gateway)
             self._shell.user_ns["context"] = context
@@ -148,6 +153,7 @@ class SQLMeshMagics(Magics):
             if args.debug:
                 logger.exception("Failed to initialize SQLMesh context")
             raise
+
         context.console.log_success(f"SQLMesh project context set to: {', '.join(args.paths)}")
 
     @magic_arguments()
diff --git a/sqlmesh/utils/pydantic.py b/sqlmesh/utils/pydantic.py
index cfe3ac04..3de15773 100644
--- a/sqlmesh/utils/pydantic.py
+++ b/sqlmesh/utils/pydantic.py
@@ -221,7 +221,8 @@ def _formatted_validation_errors(error: pydantic.ValidationError) -> t.List[str]
     for e in error.errors():
         msg = e["msg"]
         loc: t.Optional[t.Tuple] = e.get("loc")
-        result.append(f"Invalid field '{loc[0]}':\n    {msg}" if loc else msg)
+        loc_str = ".".join(loc) if loc else None
+        result.append(f"Invalid field '{loc_str}':\n    {msg}" if loc_str else msg)
     return result
 
 
@@ -305,6 +306,24 @@ def cron_validator(v: t.Any) -> str:
     return v
 
 
+def get_concrete_types_from_typehint(typehint: type[t.Any]) -> set[type[t.Any]]:
+    concrete_types = set()
+    unpacked = t.get_origin(typehint)
+    if unpacked is None:
+        if type(typehint) == type(type):
+            return {typehint}
+    elif unpacked is t.Union:
+        for item in t.get_args(typehint):
+            if str(item).startswith("typing."):
+                concrete_types |= get_concrete_types_from_typehint(item)
+            else:
+                concrete_types.add(item)
+    else:
+        concrete_types.add(unpacked)
+
+    return concrete_types
+
+
 if t.TYPE_CHECKING:
     SQLGlotListOfStrings = t.List[str]
     SQLGlotString = str
diff --git a/vscode/extension/package.json b/vscode/extension/package.json
index fd86d7cd..536b9e12 100644
--- a/vscode/extension/package.json
+++ b/vscode/extension/package.json
@@ -27,6 +27,17 @@
     "ms-python.python"
   ],
   "contributes": {
+    "configuration": {
+      "type": "object",
+      "title": "SQLMesh",
+      "properties": {
+        "sqlmesh.projectPath": {
+          "type": "string",
+          "default": "",
+          "markdownDescription": "The path to the SQLMesh project. If not set, the extension will try to find the project root automatically. If set, the extension will use the project root as the workspace path, e.g. it will run `sqlmesh` and `sqlmesh_lsp` in the project root. The path can be absolute `/Users/sqlmesh_user/sqlmesh_project/sushi` or relative `./project_folder/sushi` to the workspace root."
+        }
+      }
+    },
     "viewsContainers": {
       "panel": [
         {
diff --git a/vscode/extension/src/lsp/lsp.ts b/vscode/extension/src/lsp/lsp.ts
index 1efa03aa..50129406 100644
--- a/vscode/extension/src/lsp/lsp.ts
+++ b/vscode/extension/src/lsp/lsp.ts
@@ -8,7 +8,7 @@ import {
 import { sqlmeshLspExec } from '../utilities/sqlmesh/sqlmesh'
 import { err, isErr, ok, Result } from '@bus/result'
 import { getWorkspaceFolders } from '../utilities/common/vscodeapi'
-import { traceError } from '../utilities/common/log'
+import { traceError, traceInfo } from '../utilities/common/log'
 import { ErrorType } from '../utilities/errors'
 import { CustomLSPMethods } from './custom'
 
@@ -43,9 +43,7 @@ export class LSPClient implements Disposable {
         message: 'Invalid number of workspace folders',
       })
     }
-
-    const folder = workspaceFolders[0]
-    const workspacePath = workspaceFolders[0].uri.fsPath
+    const workspacePath = sqlmesh.value.workspacePath
     const serverOptions: ServerOptions = {
       run: {
         command: sqlmesh.value.bin,
@@ -66,11 +64,13 @@ export class LSPClient implements Disposable {
     }
     const clientOptions: LanguageClientOptions = {
       documentSelector: [{ scheme: 'file', pattern: `**/*.sql` }],
-      workspaceFolder: folder,
       diagnosticCollectionName: 'sqlmesh',
       outputChannel: outputChannel,
     }
 
+    traceInfo(
+      `Starting SQLMesh Language Server with workspace path: ${workspacePath} with server options ${JSON.stringify(serverOptions)} and client options ${JSON.stringify(clientOptions)}`,
+    )
     this.client = new LanguageClient(
       'sqlmesh-lsp',
       'SQLMesh Language Server',
diff --git a/vscode/extension/src/utilities/config.ts b/vscode/extension/src/utilities/config.ts
new file mode 100644
index 00000000..e77a39ce
--- /dev/null
+++ b/vscode/extension/src/utilities/config.ts
@@ -0,0 +1,77 @@
+import { workspace, WorkspaceFolder } from 'vscode'
+import path from 'path'
+import fs from 'fs'
+import { Result, err, ok } from '@bus/result'
+import { traceVerbose, traceInfo } from './common/log'
+
+export interface SqlmeshConfiguration {
+  projectPath: string
+}
+
+/**
+ * Get the SQLMesh configuration from VS Code settings.
+ *
+ * @returns The SQLMesh configuration
+ */
+export function getSqlmeshConfiguration(): SqlmeshConfiguration {
+  const config = workspace.getConfiguration('sqlmesh')
+  const projectPath = config.get<string>('projectPath', '')
+  return {
+    projectPath,
+  }
+}
+
+/**
+ * Validate and resolve the project path from configuration.
+ * If no project path is configured, use the workspace folder.
+ * If the project path is configured, it must be a directory that contains a SQLMesh project.
+ *
+ * @param workspaceFolder The current workspace folder
+ * @returns A Result containing the resolved project path or an error
+ */
+export function resolveProjectPath(
+  workspaceFolder: WorkspaceFolder,
+): Result<string, string> {
+  const config = getSqlmeshConfiguration()
+
+  if (!config.projectPath) {
+    // If no project path is configured, use the workspace folder
+    traceVerbose('No project path configured, using workspace folder')
+    return ok(workspaceFolder.uri.fsPath)
+  }
+  let resolvedPath: string
+
+  // Check if the path is absolute
+  if (path.isAbsolute(config.projectPath)) {
+    resolvedPath = config.projectPath
+  } else {
+    // Resolve relative path from workspace root
+    resolvedPath = path.join(workspaceFolder.uri.fsPath, config.projectPath)
+  }
+
+  // Normalize the path
+  resolvedPath = path.normalize(resolvedPath)
+
+  // Validate that the path exists
+  if (!fs.existsSync(resolvedPath)) {
+    return err(`Configured project path does not exist: ${resolvedPath}`)
+  }
+
+  // Validate that it's a directory
+  const stats = fs.statSync(resolvedPath)
+  if (!stats.isDirectory()) {
+    return err(`Configured project path is not a directory: ${resolvedPath}`)
+  }
+
+  // Check if it contains SQLMesh project files (config.yaml, config.yml, or config.py)
+  const configFiles = ['config.yaml', 'config.yml', 'config.py']
+  const hasConfigFile = configFiles.some(file =>
+    fs.existsSync(path.join(resolvedPath, file)),
+  )
+  if (!hasConfigFile) {
+    traceInfo(`Warning: No SQLMesh configuration file found in ${resolvedPath}`)
+  }
+
+  traceVerbose(`Using project path: ${resolvedPath}`)
+  return ok(resolvedPath)
+}
diff --git a/vscode/extension/src/utilities/sqlmesh/sqlmesh.ts b/vscode/extension/src/utilities/sqlmesh/sqlmesh.ts
index aa11033c..6adb0fa1 100644
--- a/vscode/extension/src/utilities/sqlmesh/sqlmesh.ts
+++ b/vscode/extension/src/utilities/sqlmesh/sqlmesh.ts
@@ -11,6 +11,7 @@ import { execAsync } from '../exec'
 import z from 'zod'
 import { ProgressLocation, window } from 'vscode'
 import { IS_WINDOWS } from '../isWindows'
+import { resolveProjectPath } from '../config'
 
 export interface SqlmeshExecInfo {
   workspacePath: string
@@ -28,8 +29,12 @@ export interface SqlmeshExecInfo {
  */
 export const isTcloudProject = async (): Promise<Result<boolean, string>> => {
   const projectRoot = await getProjectRoot()
-  const tcloudYamlPath = path.join(projectRoot.uri.fsPath, 'tcloud.yaml')
-  const tcloudYmlPath = path.join(projectRoot.uri.fsPath, 'tcloud.yml')
+  const resolvedPath = resolveProjectPath(projectRoot)
+  if (isErr(resolvedPath)) {
+    return err(resolvedPath.error)
+  }
+  const tcloudYamlPath = path.join(resolvedPath.value, 'tcloud.yaml')
+  const tcloudYmlPath = path.join(resolvedPath.value, 'tcloud.yml')
   const isTcloudYamlFilePresent = fs.existsSync(tcloudYamlPath)
   const isTcloudYmlFilePresent = fs.existsSync(tcloudYmlPath)
   if (isTcloudYamlFilePresent || isTcloudYmlFilePresent) {
@@ -83,8 +88,15 @@ export const isSqlmeshEnterpriseInstalled = async (): Promise<
     return tcloudBin
   }
   const projectRoot = await getProjectRoot()
+  const resolvedPath = resolveProjectPath(projectRoot)
+  if (isErr(resolvedPath)) {
+    return err({
+      type: 'generic',
+      message: resolvedPath.error,
+    })
+  }
   const called = await execAsync(tcloudBin.value, ['is_sqlmesh_installed'], {
-    cwd: projectRoot.uri.fsPath,
+    cwd: resolvedPath.value,
   })
   if (called.exitCode !== 0) {
     return err({
@@ -183,7 +195,14 @@ export const sqlmeshExec = async (): Promise<
 > => {
   const sqlmesh = IS_WINDOWS ? 'sqlmesh.exe' : 'sqlmesh'
   const projectRoot = await getProjectRoot()
-  const workspacePath = projectRoot.uri.fsPath
+  const resolvedPath = resolveProjectPath(projectRoot)
+  if (isErr(resolvedPath)) {
+    return err({
+      type: 'generic',
+      message: resolvedPath.error,
+    })
+  }
+  const workspacePath = resolvedPath.value
   const interpreterDetails = await getInterpreterDetails()
   traceLog(`Interpreter details: ${JSON.stringify(interpreterDetails)}`)
   if (interpreterDetails.path) {
@@ -300,7 +319,14 @@ export const sqlmeshLspExec = async (): Promise<
 > => {
   const sqlmeshLSP = IS_WINDOWS ? 'sqlmesh_lsp.exe' : 'sqlmesh_lsp'
   const projectRoot = await getProjectRoot()
-  const workspacePath = projectRoot.uri.fsPath
+  const resolvedPath = resolveProjectPath(projectRoot)
+  if (isErr(resolvedPath)) {
+    return err({
+      type: 'generic',
+      message: resolvedPath.error,
+    })
+  }
+  const workspacePath = resolvedPath.value
   const interpreterDetails = await getInterpreterDetails()
   traceLog(`Interpreter details: ${JSON.stringify(interpreterDetails)}`)
   if (interpreterDetails.path) {


Test patch:
diff --git a/pytest.ini b/pytest.ini
deleted file mode 100644
index d8135996..00000000
--- a/pytest.ini
+++ /dev/null
@@ -1,45 +0,0 @@
-[pytest]
-markers =
-    # Test Type Markers
-    # Tests are ordered from fastest to slowest
-    fast: fast tests (automatically applied if no type markers)
-    slow: slow tests that typically involve interacting with a local DB (like DuckDB)
-    docker: test that involves interacting with a Docker container
-    remote: test that involves interacting with a remote DB
-    cicdonly: test that only runs on CI/CD
-    isolated: tests that need to run sequentially usually because they use fork
-
-    # Test Domain Markers
-    # default: core functionality
-    cli: test for CLI
-    dbt: test for dbt adapter
-    github: test for Github CI/CD bot
-    jupyter: tests for Jupyter integration
-    web: tests for web UI
-    # Engine Adapters
-    engine: test all engine adapters
-    athena: test for Athena
-    bigquery: test for BigQuery
-    clickhouse: test for Clickhouse (standalone mode / cluster mode)
-    clickhouse_cloud: test for Clickhouse (cloud mode)
-    databricks: test for Databricks
-    duckdb: test for DuckDB
-    motherduck: test for MotherDuck
-    mssql: test for MSSQL
-    mysql: test for MySQL
-    postgres: test for Postgres
-    redshift: test for Redshift
-    snowflake: test for Snowflake
-    spark: test for Spark
-    trino: test for Trino (all connectors)
-    risingwave: test for Risingwave
-addopts = -n 0 --dist=loadgroup
-
-asyncio_default_fixture_loop_scope = session
-
-# Set this to True to enable logging during tests
-log_cli = False
-log_cli_format = %(asctime)s.%(msecs)03d %(filename)s:%(lineno)d %(levelname)s %(message)s
-log_cli_level = INFO
-filterwarnings =
-    ignore:The localize method is no longer necessary, as this time zone supports the fold attribute
diff --git a/tests/core/engine_adapter/integration/conftest.py b/tests/core/engine_adapter/integration/conftest.py
index 35724af1..f072ca77 100644
--- a/tests/core/engine_adapter/integration/conftest.py
+++ b/tests/core/engine_adapter/integration/conftest.py
@@ -106,25 +106,23 @@ def create_test_context(
             is_remote=is_remote,
         )
 
-        skip = False
         try:
             ctx.init()
-        except Exception:
-            # We need to catch this exception because if there is an error during setup, pytest-retry aborts immediately
-            # instead of retrying
+        except:
+            # pytest-retry doesnt work if there are errors in fixture setup (ref: https://github.com/str0zzapreti/pytest-retry/issues/33 )
+            # what we can do is log the exception and return a partially-initialized context to the test, which should
+            # throw an exception when it tries to access something that didnt init properly and thus trigger pytest-retry to retry
             logger.exception("Context init failed")
-            skip = True
-
-        if not skip:
-            with ctx.engine_adapter.session({}):
-                yield ctx
-
-            try:
-                ctx.cleanup()
-            except Exception:
-                # We need to catch this exception because if there is an error during teardown, pytest-retry aborts immediately
-                # instead of retrying
-                logger.exception("Context cleanup failed")
+
+        with ctx.engine_adapter.session({}):
+            yield ctx
+
+        try:
+            ctx.cleanup()
+        except:
+            # We need to catch this exception because if there is an error during teardown, pytest-retry aborts immediately
+            # instead of retrying
+            logger.exception("Context cleanup failed")
 
     return _create
 
diff --git a/tests/core/engine_adapter/integration/docker/compose.trino.yaml b/tests/core/engine_adapter/integration/docker/compose.trino.yaml
index a3d904b8..f5ae25fa 100644
--- a/tests/core/engine_adapter/integration/docker/compose.trino.yaml
+++ b/tests/core/engine_adapter/integration/docker/compose.trino.yaml
@@ -28,7 +28,7 @@ services:
 
   # Trino Stack
   trino:
-    image: 'trinodb/trino:465'
+    image: 'trinodb/trino:475'
     ports:
       - '8080:8080'
     volumes:
diff --git a/tests/core/state_sync/test_state_sync.py b/tests/core/state_sync/test_state_sync.py
index 1129420b..f3937f48 100644
--- a/tests/core/state_sync/test_state_sync.py
+++ b/tests/core/state_sync/test_state_sync.py
@@ -949,7 +949,7 @@ def test_promote_snapshots_no_gaps(state_sync: EngineAdapterStateSync, make_snap
     state_sync.add_interval(new_snapshot_missing_interval, "2022-01-01", "2022-01-02")
     with pytest.raises(
         SQLMeshError,
-        match=r'Detected missing intervals for model "a", interrupting your current plan. Please re-apply your plan to resolve this error.',
+        match=r".*Detected missing intervals for model .*, interrupting your current plan. Please re-apply your plan to resolve this error.*",
     ):
         promote_snapshots(state_sync, [new_snapshot_missing_interval], "prod", no_gaps=True)
 
@@ -1061,7 +1061,7 @@ def test_start_date_gap(state_sync: EngineAdapterStateSync, make_snapshot: t.Cal
     state_sync.add_interval(snapshot, "2022-01-03", "2022-01-04")
     with pytest.raises(
         SQLMeshError,
-        match=r'Detected missing intervals for model "a", interrupting your current plan. Please re-apply your plan to resolve this error.',
+        match=r".*Detected missing intervals for model .*, interrupting your current plan. Please re-apply your plan to resolve this error.*",
     ):
         promote_snapshots(state_sync, [snapshot], "prod", no_gaps=True)
 
diff --git a/tests/core/test_config.py b/tests/core/test_config.py
index e580ed94..d93a5fb1 100644
--- a/tests/core/test_config.py
+++ b/tests/core/test_config.py
@@ -985,3 +985,37 @@ def test_config_subclassing() -> None:
     class ConfigSubclass(Config): ...
 
     ConfigSubclass()
+
+
+def test_config_complex_types_supplied_as_json_strings_from_env(tmp_path: Path) -> None:
+    config_path = tmp_path / "config.yaml"
+    config_path.write_text("""
+    gateways:
+      bigquery:
+        connection:
+          type: bigquery
+          project: unit-test
+
+    default_gateway: bigquery
+
+    model_defaults:
+      dialect: bigquery
+""")
+    with mock.patch.dict(
+        os.environ,
+        {
+            "SQLMESH__GATEWAYS__BIGQUERY__CONNECTION__SCOPES": '     ["a","b","c"]',  # note: leading whitespace is deliberate
+            "SQLMESH__GATEWAYS__BIGQUERY__CONNECTION__KEYFILE_JSON": '{ "foo": "bar" }',
+        },
+    ):
+        config = load_config_from_paths(
+            Config,
+            project_paths=[config_path],
+        )
+
+        conn = config.gateways["bigquery"].connection
+        assert isinstance(conn, BigQueryConnectionConfig)
+
+        assert conn.project == "unit-test"
+        assert conn.scopes == ("a", "b", "c")
+        assert conn.keyfile_json == {"foo": "bar"}
diff --git a/tests/core/test_connection_config.py b/tests/core/test_connection_config.py
index 8680e094..f30aba11 100644
--- a/tests/core/test_connection_config.py
+++ b/tests/core/test_connection_config.py
@@ -617,6 +617,27 @@ def test_duckdb_attach_options():
     assert options.to_sql(alias="db") == "ATTACH IF NOT EXISTS 'test.db' AS db"
 
 
+def test_duckdb_config_json_strings(make_config):
+    config = make_config(
+        type="duckdb",
+        extensions='["foo","bar"]',
+        catalogs="""{
+            "test1": "test1.duckdb",
+            "test2": {
+                "type": "duckdb",
+                "path": "test2.duckdb"
+            }
+        }""",
+    )
+    assert isinstance(config, DuckDBConnectionConfig)
+
+    assert config.extensions == ["foo", "bar"]
+
+    assert config.get_catalog() == "test1"
+    assert config.catalogs.get("test1") == "test1.duckdb"
+    assert config.catalogs.get("test2").path == "test2.duckdb"
+
+
 def test_motherduck_attach_catalog(make_config):
     config = make_config(
         type="motherduck",
@@ -779,6 +800,21 @@ def test_bigquery(make_config):
         make_config(type="bigquery", quota_project="quota_project", check_import=False)
 
 
+def test_bigquery_config_json_string(make_config):
+    config = make_config(
+        type="bigquery",
+        project="project",
+        # these can be present as strings if they came from env vars
+        scopes='["a","b","c"]',
+        keyfile_json='{"foo":"bar"}',
+    )
+
+    assert isinstance(config, BigQueryConnectionConfig)
+
+    assert config.scopes == ("a", "b", "c")
+    assert config.keyfile_json == {"foo": "bar"}
+
+
 def test_postgres(make_config):
     config = make_config(
         type="postgres",
@@ -1016,7 +1052,10 @@ def test_engine_import_validator():
     with pytest.raises(
         ConfigError,
         match=re.escape(
-            """Failed to import the 'bigquery' engine library. Please run `pip install "sqlmesh[bigquery]"`."""
+            "Failed to import the 'bigquery' engine library. This may be due to a missing "
+            "or incompatible installation. Please ensure the required dependency is installed by "
+            'running: `pip install "sqlmesh[bigquery]"`. For more details, check the logs '
+            "in the 'logs/' folder, or rerun the command with the '--debug' flag."
         ),
     ):
 
@@ -1028,7 +1067,10 @@ def test_engine_import_validator():
     with pytest.raises(
         ConfigError,
         match=re.escape(
-            """Failed to import the 'bigquery' engine library. Please run `pip install "sqlmesh[bigquery_extra]"`."""
+            "Failed to import the 'bigquery' engine library. This may be due to a missing "
+            "or incompatible installation. Please ensure the required dependency is installed by "
+            'running: `pip install "sqlmesh[bigquery_extra]"`. For more details, check the logs '
+            "in the 'logs/' folder, or rerun the command with the '--debug' flag."
         ),
     ):
 
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index a6607a74..73524274 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -4919,6 +4919,10 @@ def test_signals():
     model = load_sql_based_model(expressions)
     assert model.signals[0][1] == {"arg": exp.Literal.number(1)}
 
+    @signal()
+    def my_signal(batch):
+        return True
+
     expressions = d.parse(
         """
         MODEL (
@@ -4946,7 +4950,10 @@ def test_signals():
         """
     )
 
-    model = load_sql_based_model(expressions)
+    model = load_sql_based_model(
+        expressions,
+        signal_definitions={"my_signal": signal.get_registry()["my_signal"]},
+    )
     assert model.signals == [
         (
             "my_signal",
@@ -9837,6 +9844,22 @@ def test_invalid_audit_reference():
         load_sql_based_model(expressions)
 
 
+def test_invalid_signal_reference():
+    sql = """
+    MODEL (
+      name test,
+      signals (s())
+    );
+
+    SELECT
+      1 AS id
+    """
+    expressions = d.parse(sql)
+
+    with pytest.raises(ConfigError, match="Signal 's' is undefined"):
+        load_sql_based_model(expressions)
+
+
 def test_scd_time_data_type_does_not_cause_diff_after_deserialization() -> None:
     for dialect in (
         "athena",
@@ -9970,6 +9993,7 @@ def test_extract_schema_in_post_statement(tmp_path: Path) -> None:
         SELECT c FROM x;
         ON_VIRTUAL_UPDATE_BEGIN;
         @check_schema('y');
+        @check_self_schema();
         ON_VIRTUAL_UPDATE_END;
         """
     )
@@ -9984,6 +10008,11 @@ from sqlmesh import macro
 def check_schema(evaluator, model_name: str):
     if evaluator.runtime_stage != 'loading':
         assert evaluator.columns_to_types(model_name) == {"c": exp.DataType.build("INT")}
+
+@macro()
+def check_self_schema(evaluator):
+    if evaluator.runtime_stage != 'loading':
+        assert evaluator.columns_to_types(evaluator.this_model_fqn) == {"c": exp.DataType.build("INT")}
 """)
 
     context = Context(paths=tmp_path, config=config)
diff --git a/tests/utils/test_pydantic.py b/tests/utils/test_pydantic.py
index 9246d5ed..b07d45ac 100644
--- a/tests/utils/test_pydantic.py
+++ b/tests/utils/test_pydantic.py
@@ -1,7 +1,9 @@
+import typing as t
+import pytest
 from functools import cached_property
 
 from sqlmesh.utils.date import TimeLike, to_date, to_datetime
-from sqlmesh.utils.pydantic import PydanticModel
+from sqlmesh.utils.pydantic import PydanticModel, get_concrete_types_from_typehint
 
 
 def test_datetime_date_serialization() -> None:
@@ -62,3 +64,26 @@ def test_pydantic_dict_default_args_override() -> None:
         name: str
 
     assert TestModel(name="foo").dict(by_alias=True)
+
+
+@pytest.mark.parametrize(
+    "input,output",
+    [
+        (t.Dict[str, t.Any], {dict}),
+        (dict, {dict}),
+        (t.List[str], {list}),
+        (list, {list}),
+        (t.Tuple[str, ...], {tuple}),
+        (tuple, {tuple}),
+        (t.Set[str], {set}),
+        (set, {set}),
+        (t.Optional[t.Dict[str, t.Any]], {dict, type(None)}),
+        (t.Optional[t.List[str]], {list, type(None)}),
+        (
+            t.Union[str, t.List[str], t.Dict[str, t.Any], t.Optional[t.Set[str]]],
+            {str, list, dict, set, type(None)},
+        ),
+    ],
+)
+def test_get_concrete_types_from_typehint(input: t.Any, output: set[type]) -> None:
+    assert get_concrete_types_from_typehint(input) == output
diff --git a/vscode/extension/tests/lineage.spec.ts b/vscode/extension/tests/lineage.spec.ts
index 2a679eed..5bac457e 100644
--- a/vscode/extension/tests/lineage.spec.ts
+++ b/vscode/extension/tests/lineage.spec.ts
@@ -1,52 +1,165 @@
-import { test, _electron as electron, expect } from '@playwright/test';
+import { test, _electron as electron, expect, ElectronApplication, Page } from '@playwright/test';
 import path from 'path';
 import fs from 'fs-extra';
+import os from 'os';
 
 // Absolute path to the VS Code executable you downloaded in step 1.
 const VS_CODE_EXE = fs.readJsonSync('.vscode-test/paths.json').executablePath;
 // Where your extension lives on disk
-const EXT_PATH = path.join(__dirname, '..');
-// Where the sushi project lives which we open in the webview
-const PROJECT_PATH = path.join(__dirname, '..', '..', '..', 'examples', 'sushi');
+const EXT_PATH = path.resolve(__dirname, '..');
+// Where the sushi project lives which we copy from
+const SUSHI_SOURCE_PATH = path.join(__dirname, '..', '..', '..', 'examples', 'sushi');
 
-test('Lineage panel renders correctly', async () => {
+/**
+ * Helper function to launch VS Code and test lineage with given project path config
+ */
+async function testLineageWithProjectPath(
+  window: Page,
+): Promise<void> {
+    // Trigger lineage command
+    await window.keyboard.press(process.platform === 'darwin' ? 'Meta+Shift+P' : 'Control+Shift+P');
+    await window.keyboard.type('Lineage: Focus On View');
+    await window.keyboard.press('Enter');
+
+    // Wait for "Loaded SQLmesh Context" text to appear
+    const loadedContextText = window.locator('text=Loaded SQLMesh Context');
+    await expect(loadedContextText.first()).toBeVisible({ timeout: 10_000 });
+}
+
+/**
+ * Launch VS Code and return the window and a function to close the app.
+ * @param workspaceDir The workspace directory to open.
+ * @returns The window and a function to close the app.
+ */
+export const startVSCode = async (workspaceDir: string): Promise<{
+  window: Page,
+  close: () => Promise<void>,
+}> => {
+  const userDataDir = await fs.mkdtemp(path.join(os.tmpdir(), 'vscode-user-data-'));
   const ciArgs = process.env.CI ? [
     '--disable-gpu',
     '--headless',
     '--no-sandbox',
-    '--disable-dev-shm-usage',  // Prevents memory issues in Docker/CI
-    '--window-position=-10000,0', // Ensures window is off-screen
+    '--disable-dev-shm-usage',
+    '--window-position=-10000,0',
   ] : [];
   const args = [
     ...ciArgs,
     `--extensionDevelopmentPath=${EXT_PATH}`,
-    '--disable-workspace-trust',          // no modal prompt
+    '--disable-workspace-trust',
     '--disable-telemetry',
-    '--user-data-dir=/tmp/vscode-test',   // throwaway profile
-    `${PROJECT_PATH}`,
+    `--user-data-dir=${userDataDir}`,
+    workspaceDir,
   ];
   const electronApp = await electron.launch({
     executablePath: VS_CODE_EXE,
     args,
   });
-
-  // ➋ Grab the first window that appears (the Workbench)
   const window = await electronApp.firstWindow();
-
-  // Wait for VS Code to be ready
   await window.waitForLoadState('domcontentloaded');
   await window.waitForLoadState('networkidle');
+  await window.waitForTimeout(2_000);
+  return { window, close: async () => {
+    await electronApp.close();
+    await fs.remove(userDataDir);
+  } };
+}
+
+test('Lineage panel renders correctly - no project path config (default)', async () => {
+  const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), 'vscode-test-sushi-'));
+  await fs.copy(SUSHI_SOURCE_PATH, tempDir);
+  try {
+  const { window, close } = await startVSCode(tempDir);
+  await testLineageWithProjectPath(window);
+    await close();
+  } finally { 
+    await fs.remove(tempDir);
+  }
+});
+
+test('Lineage panel renders correctly - relative project path', async () => {
+  const workspaceDir = await fs.mkdtemp(path.join(os.tmpdir(), 'vscode-test-workspace-'));
+  const projectDir = path.join(workspaceDir, 'projects', 'sushi');
+  await fs.copy(SUSHI_SOURCE_PATH, projectDir);
+
+  const settings = {
+    "sqlmesh.projectPath": "./projects/sushi",
+  };
+  await fs.ensureDir(path.join(workspaceDir, '.vscode'));
+  await fs.writeJson(path.join(workspaceDir, '.vscode', 'settings.json'), settings, { spaces: 2 });
+
+  try {
+    const { window, close } = await startVSCode(workspaceDir);
+    await testLineageWithProjectPath(window);
+    await close();
+  } finally {
+    await fs.remove(workspaceDir);
+  }
+});
+
+test('Lineage panel renders correctly - absolute project path', async () => {
+  const workspaceDir = await fs.mkdtemp(path.join(os.tmpdir(), 'vscode-test-workspace-'));
+  const projectDir = path.join(workspaceDir, 'projects', 'sushi');
+  await fs.ensureDir(path.join(workspaceDir, '.vscode'));
+  await fs.copy(SUSHI_SOURCE_PATH, projectDir);
+  await fs.ensureDir(path.join(workspaceDir, '.vscode'));
+  const settings = {
+    "sqlmesh.projectPath": projectDir,
+  };
+  await fs.writeJson(path.join(workspaceDir, '.vscode', 'settings.json'), settings, { spaces: 2 });
+
+  try {
+    const { window, close } = await startVSCode(workspaceDir);
+    await testLineageWithProjectPath(window);
+    await close();
+  } finally {
+    await fs.remove(workspaceDir);
+  }
+});
+
 
-  await expect(window.locator("text=lineage").first()).toBeVisible();
+test("Lineage panel renders correctly - relative project outside of workspace", async () => {
+  const tempFolder = await fs.mkdtemp(path.join(os.tmpdir(), 'vscode-test-workspace-'));
+  const projectDir = path.join(tempFolder, 'projects', 'sushi');
+  await fs.copy(SUSHI_SOURCE_PATH, projectDir);
 
-  // ➌ Trigger our command exactly like a user would
-  await window.keyboard.press(process.platform === 'darwin' ? 'Meta+Shift+P' : 'Control+Shift+P');
-  await window.keyboard.type('Lineage: Focus On View');
-  await window.keyboard.press('Enter');
+  const workspaceDir = path.join(tempFolder, 'workspace');
+  await fs.ensureDir(workspaceDir);
 
-  // Wait for "Loaded SQLmesh Context" text to appear
-  const loadedContextText = window.locator('text=Loaded SQLmesh Context');
-  await expect(loadedContextText).toBeVisible({ timeout: 10000 });
+  const settings = {
+    "sqlmesh.projectPath": "./../projects/sushi",
+  };
+  await fs.ensureDir(path.join(workspaceDir, '.vscode'));
+  await fs.writeJson(path.join(workspaceDir, '.vscode', 'settings.json'), settings, { spaces: 2 });
 
-  await electronApp.close();
+  try {
+    const { window, close } = await startVSCode(workspaceDir);
+    await testLineageWithProjectPath(window);
+    await close();
+  } finally {
+    await fs.remove(tempFolder);
+  }
 });
+
+test("Lineage panel renders correctly - absolute path project outside of workspace", async () => {
+  const tempFolder = await fs.mkdtemp(path.join(os.tmpdir(), 'vscode-test-workspace-'));
+  const projectDir = path.join(tempFolder, 'projects', 'sushi');
+  await fs.copy(SUSHI_SOURCE_PATH, projectDir);
+
+  const workspaceDir = path.join(tempFolder, 'workspace');
+  await fs.ensureDir(workspaceDir);
+
+  const settings = {
+    "sqlmesh.projectPath": projectDir,
+  };
+  await fs.ensureDir(path.join(workspaceDir, '.vscode'));
+  await fs.writeJson(path.join(workspaceDir, '.vscode', 'settings.json'), settings, { spaces: 2 });
+
+  try {
+    const { window, close } = await startVSCode(workspaceDir);
+    await testLineageWithProjectPath(window);
+    await close();
+  } finally {
+    await fs.remove(tempFolder);
+  }
+});
\ No newline at end of file