diff --git a/pyproject.toml b/pyproject.toml
index 4659aa5..da13c09 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -35,6 +35,7 @@ dependencies = [
     "httpx==0.28.1",
     "pyyaml==6.0.3",
     "pydantic==2.13.3",
+    "ruamel-yaml==0.19.1",
 ]
 
 [project.optional-dependencies]
diff --git a/src/tycoon/commands/explore.py b/src/tycoon/commands/explore.py
index 94d9153..4bfde8c 100644
--- a/src/tycoon/commands/explore.py
+++ b/src/tycoon/commands/explore.py
@@ -8,7 +8,7 @@
 import click
 import typer
 
-from tycoon.config import config
+from tycoon.config import TycoonConfig, load_config
 from tycoon.utils.console import error, header, info, success, warn
 
 
@@ -75,12 +75,13 @@ def analyze_cmd(
     from tycoon.utils.duckdb_utils import get_tables
 
     # 1. Verify tycoon.yml exists
-    if not config.has_project_file:
+    cfg = load_config()
+    if not cfg.has_project_file:
         error("No tycoon.yml found. Run 'tycoon init' first.")
         raise typer.Exit(1)
 
     # 2. Resolve which source(s) we're analyzing.
-    sources = config.sources
+    sources = cfg.sources
     if all_sources:
         if source_name:
             error("Pass either a source name or --all, not both.")
@@ -88,7 +89,7 @@ def analyze_cmd(
         if not sources:
             error("No sources registered in tycoon.yml. Run 'tycoon data sources add' first.")
             raise typer.Exit(1)
-        _analyze_all(force=force, no_dbt=no_dbt, rill=rill, build=build)
+        _analyze_all(cfg=cfg, force=force, no_dbt=no_dbt, rill=rill, build=build)
         return
 
     if not source_name:
@@ -112,7 +113,7 @@ def analyze_cmd(
     info(f"Schema: {schema_name}")
 
     # 3. Verify raw database exists and has data for this schema
-    raw_db = config.raw_db
+    raw_db = cfg.raw_db
     if not raw_db.exists():
         error(f"Raw database not found at {raw_db}. Run 'tycoon data sources run {source_name}' first.")
         raise typer.Exit(1)
@@ -133,7 +134,7 @@ def analyze_cmd(
     # 4. Generate dbt staging models
     if not no_dbt:
         info("Generating dbt staging models...")
-        staging_dir = config.dbt_project_dir / "models" / "staging" / source_name
+        staging_dir = cfg.dbt_project_dir / "models" / "staging" / source_name
         try:
             result = generate_staging_models(
                 raw_db_path=raw_db,
@@ -167,7 +168,7 @@ def analyze_cmd(
         from tycoon.scaffolding.rill_generator import generate_rill_config
         from tycoon.scaffolding.templates import scaffold_rill_dir
 
-        rill_dir = config.rill_dir
+        rill_dir = cfg.rill_dir
         if not rill_dir.exists():
             info(f"Rill project not found; scaffolding at {rill_dir}")
             scaffold_rill_dir(rill_dir)
@@ -222,7 +223,7 @@ def analyze_cmd(
             success("dbt build completed successfully.")
 
 
-def _analyze_all(*, force: bool, no_dbt: bool, rill: bool, build: bool) -> None:
+def _analyze_all(*, cfg: TycoonConfig, force: bool, no_dbt: bool, rill: bool, build: bool) -> None:
     """Iterate every registered source and analyze each.
 
     Soft-skips sources whose raw DB doesn't exist yet. ``--rill`` and
@@ -230,7 +231,7 @@ def _analyze_all(*, force: bool, no_dbt: bool, rill: bool, build: bool) -> None:
     """
     from tycoon.scaffolding.dbt_generator import generate_staging_models
 
-    sources = config.sources
+    sources = cfg.sources
     header(f"Analyzing all sources ({len(sources)})")
 
     total_generated: list[str] = []
@@ -239,7 +240,7 @@ def _analyze_all(*, force: bool, no_dbt: bool, rill: bool, build: bool) -> None:
 
     for src_name, src_cfg in sources.items():
         info(f"  → {src_name} (schema: {src_cfg.schema_name})")
-        raw_db = config.raw_db
+        raw_db = cfg.raw_db
         if not raw_db.exists():
             warn(
                 f"    Skipping {src_name} — raw DB not found at {raw_db}. "
@@ -251,7 +252,7 @@ def _analyze_all(*, force: bool, no_dbt: bool, rill: bool, build: bool) -> None:
         if no_dbt:
             continue
 
-        staging_dir = config.dbt_project_dir / "models" / "staging" / src_name
+        staging_dir = cfg.dbt_project_dir / "models" / "staging" / src_name
         try:
             result = generate_staging_models(
                 raw_db_path=raw_db,
diff --git a/src/tycoon/commands/init.py b/src/tycoon/commands/init.py
index b922531..c06fc78 100644
--- a/src/tycoon/commands/init.py
+++ b/src/tycoon/commands/init.py
@@ -14,6 +14,7 @@
     StackConfig,
     TransformationTool,
     WarehouseType,
+    migrate_project,
 )
 from tycoon.scaffolding.templates import (
     list_templates,
@@ -504,8 +505,31 @@ def init_cmd(
             ),
         ),
     ] = None,
+    upgrade: Annotated[
+        bool,
+        typer.Option(
+            "--upgrade",
+            help="Migrate tycoon.yml to the current schema version and exit.",
+        ),
+    ] = False,
 ) -> None:
     """Initialize a new tycoon project in the current directory."""
+    if upgrade:
+        target = Path.cwd()
+        if not (target / "tycoon.yml").exists():
+            error("No tycoon.yml found in the current directory. Run 'tycoon init' to create one.")
+            raise typer.Exit(1)
+        try:
+            changed = migrate_project(target)
+        except ValueError as exc:
+            error(str(exc))
+            raise typer.Exit(1) from exc
+        if changed:
+            success("tycoon.yml migrated to the current schema version.")
+        else:
+            info("tycoon.yml is already up to date.")
+        raise typer.Exit(0)
+
     if list_templates_flag:
         templates = list_templates()
         if not templates:
diff --git a/src/tycoon/commands/run_all.py b/src/tycoon/commands/run_all.py
index 0462f4b..bea3815 100644
--- a/src/tycoon/commands/run_all.py
+++ b/src/tycoon/commands/run_all.py
@@ -8,7 +8,7 @@
 
 import typer
 
-from tycoon.config import config
+from tycoon.config import load_config
 from tycoon.utils.console import console, error, header, info, next_steps, success, warn
 
 
@@ -46,7 +46,8 @@ def run_all_cmd(
     ] = False,
 ) -> None:
     """Ingest all registered sources then run dbt build."""
-    if not config.has_project_file:
+    cfg = load_config()
+    if not cfg.has_project_file:
         error("No tycoon.yml found. Run [bold]tycoon init[/bold] first.")
         raise typer.Exit(1)
 
@@ -59,7 +60,7 @@ def _emit(severity: str, message: str, **fields: str) -> None:
             return
         from tycoon import notify as notify_mod
 
-        project = config.project
+        project = cfg.project
         prefs = project.notify if project is not None else None
         allowed = prefs.severities if prefs is not None else ["success", "error"]
         if severity not in allowed:
@@ -75,12 +76,12 @@ def _emit(severity: str, message: str, **fields: str) -> None:
     if not skip_ingest:
         from tycoon.ingestion.runner import run_source as _run_source
 
-        sources = config.sources
+        sources = cfg.sources
         if not sources:
             error("No sources registered. Run [bold]tycoon data sources add[/bold] first.")
             raise typer.Exit(1)
 
-        config.ensure_data_dir()
+        cfg.ensure_data_dir()
         total = len(sources)
         info(f"Ingesting {total} source{'s' if total != 1 else ''}...")
         if max_records is not None:
@@ -92,7 +93,7 @@ def _emit(severity: str, message: str, **fields: str) -> None:
                 _pipeline, load_info = _run_source(
                     name=name,
                     source_config=source_config,
-                    raw_db_path=config.raw_db,
+                    raw_db_path=cfg.raw_db,
                     max_records=max_records,
                 )
                 success(f"{name}: {load_info}")
@@ -111,7 +112,7 @@ def _emit(severity: str, message: str, **fields: str) -> None:
             _emit("error", "run-all failed: dbt not found on PATH", stage="transform")
             raise typer.Exit(1)
 
-        project_dir = config.dbt_project_dir
+        project_dir = cfg.dbt_project_dir
         if not project_dir.exists():
             warn(f"dbt project not found at {project_dir} — skipping transform.")
         else:
@@ -132,7 +133,7 @@ def _emit(severity: str, message: str, **fields: str) -> None:
     elapsed = time.time() - start
     console.rule("[bold green]Done")
     success(f"Finished in {elapsed:.1f}s")
-    _emit("success", "run-all complete", elapsed=f"{elapsed:.1f}s", sources=str(len(config.sources)))
+    _emit("success", "run-all complete", elapsed=f"{elapsed:.1f}s", sources=str(len(cfg.sources)))
     next_steps(
         ("tycoon data status", "check source freshness and row counts"),
         ("tycoon start --only rill", "explore results in Rill"),
diff --git a/src/tycoon/commands/sources.py b/src/tycoon/commands/sources.py
index 31fb06a..3680a5c 100644
--- a/src/tycoon/commands/sources.py
+++ b/src/tycoon/commands/sources.py
@@ -9,7 +9,7 @@
 import typer
 from rich.table import Table
 
-from tycoon.config import config
+from tycoon.config import TycoonConfig, load_config
 from tycoon.ingestion.catalog import CATALOG, CatalogEntry
 from tycoon.project import SourceConfig, load_project, save_project
 from tycoon.utils.console import console, error, header, info, next_steps, success, warn
@@ -32,11 +32,13 @@
 app.add_typer(list_app, name="list")
 
 
-def _require_project() -> None:
-    """Abort if no tycoon.yml exists."""
-    if not config.has_project_file:
+def _require_project() -> TycoonConfig:
+    """Return a fresh TycoonConfig, aborting if no tycoon.yml exists."""
+    cfg = load_config()
+    if not cfg.has_project_file:
         error("No tycoon.yml found. Run [bold]tycoon init[/bold] first.")
         raise typer.Exit(1)
+    return cfg
 
 
 # ---------------------------------------------------------------------------
@@ -79,9 +81,9 @@ def catalog_default(ctx: typer.Context) -> None:
 
 def _list_sources() -> None:
     """Print registered sources table."""
-    _require_project()
+    cfg = _require_project()
 
-    sources = config.sources
+    sources = cfg.sources
     if not sources:
         info("No sources registered yet.")
         info("Browse available sources with [bold]tycoon data sources catalog[/bold]")
@@ -110,9 +112,9 @@ def show_source(
     name: str = typer.Argument(help="Name of the source to show"),
 ) -> None:
     """Show detailed configuration for a specific source."""
-    _require_project()
+    cfg = _require_project()
 
-    sources = config.sources
+    sources = cfg.sources
     if name not in sources:
         error(f"Source [bold]{name}[/bold] not found.")
         info(f"Available sources: {', '.join(sources.keys()) if sources else '(none)'}")
@@ -381,7 +383,7 @@ def add_source(
     credentials default to ``${ENV_VAR}`` references in both modes —
     set the env var separately.
     """
-    _require_project()
+    cfg = _require_project()
 
     if not source_type:
         if no_prompt:
@@ -449,7 +451,7 @@ def add_source(
         config=source_config,
     )
 
-    project = load_project(config.root)
+    project = load_project(cfg.root)
     assert project is not None  # guarded by _require_project
 
     if source_name in project.sources:
@@ -464,8 +466,7 @@ def add_source(
                 raise typer.Exit(0)
 
     project.sources[source_name] = new_source
-    save_project(project, config.root)
-    config.reload()
+    save_project(project, cfg.root)
 
     success(f"Source [bold]{source_name}[/bold] added to tycoon.yml")
 
@@ -534,9 +535,9 @@ def remove_source(
     name: str = typer.Argument(help="Name of the source to remove"),
 ) -> None:
     """Remove a registered data source."""
-    _require_project()
+    cfg = _require_project()
 
-    project = load_project(config.root)
+    project = load_project(cfg.root)
     assert project is not None
 
     if name not in project.sources:
@@ -547,8 +548,7 @@ def remove_source(
     typer.confirm(f"Remove source '{name}'?", abort=True)
 
     del project.sources[name]
-    save_project(project, config.root)
-    config.reload()
+    save_project(project, cfg.root)
 
     success(f"Source [bold]{name}[/bold] removed from tycoon.yml")
 
@@ -585,7 +585,7 @@ def _source_already_referenced(dbt_dir: Path, source_name: str) -> bool:
     return False
 
 
-def _maybe_auto_scaffold(source_name: str, source_config: SourceConfig, *, scaffold: bool) -> None:
+def _maybe_auto_scaffold(source_name: str, source_config: SourceConfig, *, cfg: TycoonConfig, scaffold: bool) -> None:
     """Auto-run the analyze flow if a dbt project exists and no staging
     models are present for this source yet.
 
@@ -595,10 +595,10 @@ def _maybe_auto_scaffold(source_name: str, source_config: SourceConfig, *, scaff
     """
     if not scaffold:
         return
-    project = config.project
+    project = cfg.project
     if project is not None and not project.transform.auto_scaffold:
         return
-    dbt_dir = config.dbt_project_dir
+    dbt_dir = cfg.dbt_project_dir
     if not dbt_dir.exists():
         return  # No dbt project to scaffold into.
 
@@ -611,7 +611,7 @@ def _maybe_auto_scaffold(source_name: str, source_config: SourceConfig, *, scaff
         from tycoon.scaffolding.dbt_generator import generate_staging_models
 
         result = generate_staging_models(
-            raw_db_path=config.raw_db,
+            raw_db_path=cfg.raw_db,
             schema_name=source_config.schema_name,
             source_name=source_name,
             output_dir=staging_dir,
@@ -644,9 +644,9 @@ def run_source(
     """Ingest data from a registered source by name."""
     from tycoon.ingestion.runner import run_source as _run_source
 
-    _require_project()
+    cfg = _require_project()
 
-    sources = config.sources
+    sources = cfg.sources
     if not source_name:
         if not sources:
             error("No sources registered. Run 'tycoon data sources add' first.")
@@ -667,17 +667,17 @@ def run_source(
     if max_records is not None:
         info(f"Record cap: {max_records:,}")
 
-    config.ensure_data_dir()
+    cfg.ensure_data_dir()
 
     try:
         _pipeline, load_info = _run_source(
             name=source_name,
             source_config=source_config,
-            raw_db_path=config.raw_db,
+            raw_db_path=cfg.raw_db,
             max_records=max_records,
         )
         success(f"{source_name} load complete. {load_info}")
-        _maybe_auto_scaffold(source_name, source_config, scaffold=not no_scaffold)
+        _maybe_auto_scaffold(source_name, source_config, cfg=cfg, scaffold=not no_scaffold)
         next_steps(
             ("tycoon data transform run", "run dbt models on the ingested data"),
             ("tycoon start --only rill", "open the Rill dashboard"),
@@ -704,9 +704,9 @@ def run_all(
     """Run all registered source pipelines sequentially."""
     from tycoon.ingestion.runner import run_source as _run_source
 
-    _require_project()
+    cfg = _require_project()
 
-    sources = config.sources
+    sources = cfg.sources
     if not sources:
         error("No sources registered. Run 'tycoon data sources add' first.")
         raise typer.Exit(1)
@@ -716,7 +716,7 @@ def run_all(
     if max_records is not None:
         info(f"Record cap per resource: {max_records:,}")
 
-    config.ensure_data_dir()
+    cfg.ensure_data_dir()
 
     for i, (name, source_config) in enumerate(sources.items(), 1):
         info(f"Step {i}/{total} — {name} ({source_config.type})...")
@@ -724,11 +724,11 @@ def run_all(
             _pipeline, load_info = _run_source(
                 name=name,
                 source_config=source_config,
-                raw_db_path=config.raw_db,
+                raw_db_path=cfg.raw_db,
                 max_records=max_records,
             )
             success(f"{name} complete. {load_info}")
-            _maybe_auto_scaffold(name, source_config, scaffold=not no_scaffold)
+            _maybe_auto_scaffold(name, source_config, cfg=cfg, scaffold=not no_scaffold)
         except Exception as exc:
             error(f"{name} pipeline failed: {exc}")
             raise typer.Exit(1) from exc
diff --git a/src/tycoon/commands/sync_cmd.py b/src/tycoon/commands/sync_cmd.py
index 714b7dd..efad64e 100644
--- a/src/tycoon/commands/sync_cmd.py
+++ b/src/tycoon/commands/sync_cmd.py
@@ -19,7 +19,7 @@
 
 import typer
 
-from tycoon.config import config
+from tycoon.config import load_config
 from tycoon.project import SyncSourceSpec
 from tycoon.sync import sync_to_local
 from tycoon.utils.console import console, error, info, next_steps, success, warn
@@ -59,7 +59,8 @@ def sync_cmd(
     the local snapshot is intentionally allowed to go stale until you
     re-run it.
     """
-    sync_cfg = config.project.sync if config.project else None
+    cfg = load_config()
+    sync_cfg = cfg.project.sync if cfg.project else None
 
     # Resolve source specs.
     if from_:
@@ -97,7 +98,7 @@ def sync_cmd(
     # Resolve destination.
     if to is None:
         if sync_cfg and sync_cfg.to:
-            to = config.root / sync_cfg.to
+            to = cfg.root / sync_cfg.to
         else:
             error("No destination. Pass [bold]--to <path>[/bold] or add [bold]sync.to[/bold] to tycoon.yml.")
             raise typer.Exit(1)
diff --git a/src/tycoon/config.py b/src/tycoon/config.py
index 3462565..83b5e31 100644
--- a/src/tycoon/config.py
+++ b/src/tycoon/config.py
@@ -8,7 +8,8 @@
 
 from pathlib import Path
 
-from tycoon.project import PROJECT_FILENAME, TycoonProject, load_project
+from tycoon.project import PROJECT_FILENAME, SCHEMA_VERSION, TycoonProject, load_project
+from tycoon.utils.console import error as _error_console, warn as _warn_console
 
 # v0.1 defaults (used when no tycoon.yml exists)
 _DEFAULT_RAW_DB = "data/raw.duckdb"
@@ -120,5 +121,29 @@ def ensure_data_dir(self) -> None:
         self.data_dir.mkdir(parents=True, exist_ok=True)
 
 
-# Singleton
+def load_config() -> TycoonConfig:
+    """Return a TycoonConfig rooted at the nearest project directory.
+
+    Prefer this over importing _find_project_root across module boundaries.
+    Emits a console warning when tycoon.yml is at an older schema version.
+    """
+    cfg = TycoonConfig(project_root=_find_project_root())
+    if cfg.project is not None:
+        sv = cfg.project.schema_version
+        if sv is not None and sv > SCHEMA_VERSION:
+            _error_console(
+                f"tycoon.yml schema_version {sv} is newer than this tycoon supports "
+                f"({SCHEMA_VERSION}). Upgrade tycoon-cli to use this project."
+            )
+            raise SystemExit(1)
+        if sv is None or sv < SCHEMA_VERSION:
+            current = sv if sv is not None else "none"
+            _warn_console(
+                f"tycoon.yml is at schema version {current}, current is {SCHEMA_VERSION}. "
+                "Run 'tycoon init --upgrade' to migrate."
+            )
+    return cfg
+
+
+# Singleton (used by modules not yet migrated to load_config)
 config = TycoonConfig()
diff --git a/src/tycoon/project.py b/src/tycoon/project.py
index 8f6236a..6bc1854 100644
--- a/src/tycoon/project.py
+++ b/src/tycoon/project.py
@@ -6,7 +6,7 @@
 import re
 from enum import Enum
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 
 import yaml
 from pydantic import BaseModel, Field, SecretStr, field_validator
@@ -279,6 +279,25 @@ class TransformConfig(BaseModel):
     )
 
 
+class RuntimeEntry(BaseModel):
+    """One named ingestion runtime declared under ``runtimes:``."""
+
+    type: Literal["dlt-managed", "dlt-project", "fivetran", "airbyte", "estuary"]
+    # only meaningful for dlt-project; ignored for cloud-managed runtimes
+    path: str | None = None
+
+
+class MetadataConfig(BaseModel):
+    """Where tycoon stores its internal state (run history, source records).
+
+    ``backend: duckdb_file`` is the only supported backend in v0.1.x.
+    ``path`` is relative to the project root.
+    """
+
+    backend: str = "duckdb_file"
+    path: str = ".tycoon/metadata.duckdb"
+
+
 class NotifyConfig(BaseModel):
     """Optional ``notify:`` block — non-secret notification prefs (#46).
 
@@ -304,6 +323,7 @@ class TycoonProject(BaseModel):
 
     name: str = Field(default="my-project", description="Project name")
     version: str = Field(default="0.1.0", description="Project version")
+    schema_version: int | None = Field(default=None, description="Tycoon schema version (managed by tycoon)")
     database: DatabaseConfig = Field(default_factory=DatabaseConfig)
     sources: dict[str, SourceConfig] = Field(default_factory=dict, description="Registered data sources")
     dbt_project_dir: str = Field(default="dbt_project", description="Path to dbt project")
@@ -332,6 +352,21 @@ class TycoonProject(BaseModel):
         description="Defaults for transform-side commands (`data analyze`, `data sources run`).",
     )
     stack: StackConfig = Field(default_factory=StackConfig)
+    runtimes: dict[str, RuntimeEntry] = Field(
+        default_factory=dict,
+        description=(
+            "Named ingestion runtimes. Keys are arbitrary labels (e.g. 'shopify', "
+            "'custom_pipeline'); each entry declares which ingestion tool owns that "
+            "pipeline and, for dlt-project runtimes, where the project lives."
+        ),
+    )
+    metadata: MetadataConfig = Field(
+        default_factory=MetadataConfig,
+        description=(
+            "Where tycoon stores internal state (run history, source records). "
+            "Defaults to a local DuckDB file at .tycoon/metadata.duckdb."
+        ),
+    )
     notify: NotifyConfig = Field(
         default_factory=lambda: NotifyConfig(),
         description="Notification preferences for `--notify` runs and `tycoon notify`.",
@@ -358,6 +393,7 @@ def _check_path_field(cls, v: str | None) -> str | None:
 
 
 PROJECT_FILENAME = "tycoon.yml"
+SCHEMA_VERSION = 2
 
 
 def load_project(project_root: Path) -> TycoonProject | None:
@@ -385,6 +421,8 @@ def save_project(project: TycoonProject, project_root: Path) -> None:
     """
     path = project_root / PROJECT_FILENAME
     data = project.model_dump(by_alias=True, exclude_none=True, mode="json")
+    if project.schema_version is not None:
+        data["schema_version"] = project.schema_version
     if path.exists():
         existing = yaml.safe_load(path.read_text())
         if isinstance(existing, dict):
@@ -392,3 +430,55 @@ def save_project(project: TycoonProject, project_root: Path) -> None:
             if existing_meta is not None and isinstance(data.get("stack"), dict):
                 data["stack"]["ingestion_metadata"] = existing_meta
     path.write_text(yaml.dump(data, default_flow_style=False, sort_keys=False))
+
+
+def migrate_project(project_root: Path) -> bool:
+    """Upgrade tycoon.yml to SCHEMA_VERSION in place.
+
+    Adds ``metadata:`` with defaults if the key is absent, then stamps
+    ``schema_version`` when it is absent or older than SCHEMA_VERSION.
+    The user's ``version`` field is never touched. Writes back only when
+    a change is needed. Returns True if the file was modified, False if
+    already up to date (idempotent). Comments and blank lines are
+    preserved via ruamel.yaml.
+    """
+    from ruamel.yaml import YAML
+
+    path = project_root / PROJECT_FILENAME
+    if not path.exists():
+        return False
+
+    ryaml = YAML()
+    ryaml.preserve_quotes = True
+
+    with path.open() as f:
+        raw = ryaml.load(f)
+
+    if not isinstance(raw, dict):
+        return False
+
+    existing = raw.get("schema_version")
+    if existing is not None and (isinstance(existing, bool) or not isinstance(existing, int)):
+        raise ValueError(f"tycoon.yml schema_version must be an integer, got {type(existing).__name__}: {existing!r}")
+    if existing is not None and existing > SCHEMA_VERSION:
+        raise ValueError(
+            f"tycoon.yml schema_version {existing} is newer than this tycoon supports ({SCHEMA_VERSION}). "
+            "Upgrade tycoon-cli to use this project."
+        )
+
+    changed = False
+
+    if "metadata" not in raw:
+        defaults = MetadataConfig()
+        raw["metadata"] = {"backend": defaults.backend, "path": defaults.path}
+        changed = True
+
+    if existing is None or existing < SCHEMA_VERSION:
+        raw["schema_version"] = SCHEMA_VERSION
+        changed = True
+
+    if changed:
+        with path.open("w") as f:
+            ryaml.dump(raw, f)
+
+    return changed
diff --git a/src/tycoon/scaffolding/templates.py b/src/tycoon/scaffolding/templates.py
index f7441db..8b404c7 100644
--- a/src/tycoon/scaffolding/templates.py
+++ b/src/tycoon/scaffolding/templates.py
@@ -14,7 +14,7 @@
 import typer
 import yaml
 
-from tycoon.project import StackConfig
+from tycoon.project import SCHEMA_VERSION, StackConfig
 from tycoon.utils.console import info, success, warn
 
 
@@ -159,6 +159,7 @@ def scaffold_blank_project(
     project_data: dict = {
         "name": name,
         "version": "0.1.0",
+        "schema_version": SCHEMA_VERSION,
         "database": {
             "raw": raw_db_path,
             "warehouse": warehouse_db_path,
@@ -537,6 +538,9 @@ def scaffold_from_template(
         warn("tycoon.yml already exists, skipping")
     else:
         shutil.copy2(src_yml, dst_yml)
+        _raw = yaml.safe_load(dst_yml.read_text()) or {}
+        _raw["schema_version"] = SCHEMA_VERSION
+        dst_yml.write_text(yaml.dump(_raw, default_flow_style=False, sort_keys=False))
         success(f"Created tycoon.yml from template '{template_name}'")
 
     # Copy any subdirectories from the template (e.g. dbt_project/, rill/)
diff --git a/tests/test_config.py b/tests/test_config.py
index d06e956..051f2bc 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -4,7 +4,8 @@
 
 from pathlib import Path
 
-from tycoon.config import TycoonConfig
+from tycoon.config import TycoonConfig, load_config
+from tycoon.project import SCHEMA_VERSION
 
 
 class TestTycoonConfig:
@@ -36,3 +37,73 @@ def test_paths_relative_to_project_root(self, tmp_config):
         assert tmp_config.data_dir == tmp_config.root / "data"
         assert tmp_config.dbt_project_dir == tmp_config.root / "dbt_project"
         assert tmp_config.rill_dir == tmp_config.root / "rill"
+
+
+class TestLoadConfigSchemaWarning:
+    """T2-4: load_config warns via console when tycoon.yml schema_version is stale."""
+
+    def test_warns_when_schema_version_absent(self, tmp_path, monkeypatch):
+        (tmp_path / "tycoon.yml").write_text("name: old-project\n")
+        monkeypatch.chdir(tmp_path)
+
+        calls = []
+        monkeypatch.setattr("tycoon.config._warn_console", lambda msg: calls.append(msg))
+
+        load_config()
+
+        assert len(calls) == 1
+        assert "tycoon init --upgrade" in calls[0]
+
+    def test_warns_when_schema_version_old(self, tmp_path, monkeypatch):
+        (tmp_path / "tycoon.yml").write_text(f"name: old\nschema_version: {SCHEMA_VERSION - 1}\n")
+        monkeypatch.chdir(tmp_path)
+
+        calls = []
+        monkeypatch.setattr("tycoon.config._warn_console", lambda msg: calls.append(msg))
+
+        load_config()
+
+        assert len(calls) == 1
+
+    def test_no_warning_when_current(self, tmp_path, monkeypatch):
+        (tmp_path / "tycoon.yml").write_text(f"name: current\nschema_version: {SCHEMA_VERSION}\n")
+        monkeypatch.chdir(tmp_path)
+
+        calls = []
+        monkeypatch.setattr("tycoon.config._warn_console", lambda msg: calls.append(msg))
+
+        load_config()
+
+        assert calls == []
+
+    def test_no_warning_without_tycoon_yml(self, tmp_path, monkeypatch):
+        (tmp_path / "pyproject.toml").write_text('[project]\nname = "x"\n')
+        monkeypatch.chdir(tmp_path)
+
+        calls = []
+        monkeypatch.setattr("tycoon.config._warn_console", lambda msg: calls.append(msg))
+
+        load_config()
+
+        assert calls == []
+
+    def test_errors_and_exits_when_schema_version_future(self, tmp_path, monkeypatch):
+        """load_config must error and exit for a schema_version newer than SCHEMA_VERSION.
+
+        The gate lives here (not in load_project) so the import-time singleton
+        never raises and --help / init --upgrade remain reachable.
+        """
+        import pytest
+
+        (tmp_path / "tycoon.yml").write_text(f"name: future\nschema_version: {SCHEMA_VERSION + 1}\n")
+        monkeypatch.chdir(tmp_path)
+
+        errors = []
+        monkeypatch.setattr("tycoon.config._error_console", lambda msg: errors.append(msg))
+
+        with pytest.raises(SystemExit) as exc_info:
+            load_config()
+
+        assert exc_info.value.code == 1
+        assert len(errors) == 1
+        assert "newer than this tycoon supports" in errors[0]
diff --git a/tests/test_explore.py b/tests/test_explore.py
index 38bb302..79dd4ec 100644
--- a/tests/test_explore.py
+++ b/tests/test_explore.py
@@ -708,11 +708,6 @@ def test_analyze_fails_when_raw_db_missing(self, cli_runner, tmp_path, monkeypat
     def test_analyze_all_with_source_name_errors(self, cli_runner, tmp_path, monkeypatch):
         monkeypatch.chdir(tmp_path)
         (tmp_path / "tycoon.yml").write_text("name: test\nversion: 0.1.0\nsources: {}\n")
-        from tycoon.commands import explore as explore_mod
-        from tycoon.config import TycoonConfig
-
-        monkeypatch.setattr(explore_mod, "config", TycoonConfig(project_root=tmp_path))
-
         result = cli_runner.invoke(app, ["data", "analyze", "my-source", "--all"])
         assert result.exit_code != 0
         combined = (result.stdout or "") + (result.stderr or "")
@@ -747,11 +742,6 @@ def test_analyze_all_iterates_every_source(self, cli_runner, tmp_path, monkeypat
         # Need a dbt project dir so output_dir parent is valid.
         (tmp_path / "dbt_project" / "models").mkdir(parents=True)
 
-        from tycoon.commands import explore as explore_mod
-        from tycoon.config import TycoonConfig
-
-        monkeypatch.setattr(explore_mod, "config", TycoonConfig(project_root=tmp_path))
-
         result = cli_runner.invoke(app, ["data", "analyze", "--all"])
         assert result.exit_code == 0, result.stdout
 
@@ -785,12 +775,6 @@ def test_analyze_interactive_prompt_does_not_crash(self, cli_runner, tmp_path, m
         con.execute("CREATE TABLE raw_src_a.items (id INTEGER)")
         con.close()
 
-        # Reload config for the analyze command
-        from tycoon.commands import explore as explore_mod
-        from tycoon.config import TycoonConfig
-
-        monkeypatch.setattr(explore_mod, "config", TycoonConfig(project_root=tmp_path))
-
         # Supply "src_a" as the interactive choice
         result = cli_runner.invoke(app, ["data", "analyze", "--no-dbt"], input="src_a\n")
         # The key assertion: no AttributeError. Exit may be 0 or non-zero depending
diff --git a/tests/test_init.py b/tests/test_init.py
index 75f0aa6..51f4588 100644
--- a/tests/test_init.py
+++ b/tests/test_init.py
@@ -5,7 +5,7 @@
 import yaml
 
 from tycoon.cli import app
-from tycoon.project import load_project
+from tycoon.project import SCHEMA_VERSION, load_project
 from tycoon.scaffolding.templates import list_templates, scaffold_blank_project
 
 
@@ -18,6 +18,7 @@ def test_init_help_exits_zero(self, cli_runner):
         assert "--template" in result.stdout
         assert "--name" in result.stdout
         assert "--list-templates" in result.stdout
+        assert "--upgrade" in result.stdout
 
     def test_init_appears_in_top_level_help(self, cli_runner):
         result = cli_runner.invoke(app, ["--help"])
@@ -294,3 +295,44 @@ def test_unknown_param_is_warned_but_not_fatal(self, cli_runner, tmp_path, monke
         assert result.exit_code == 0
         # The unknown-param warning goes to stdout via the console helper
         assert "bogus" in result.stdout.lower() or "unknown parameter" in result.stdout.lower()
+
+
+class TestUpgrade:
+    """tycoon init --upgrade migrates tycoon.yml to the current schema version."""
+
+    def test_upgrade_no_tycoon_yml_exits_nonzero(self, cli_runner, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        result = cli_runner.invoke(app, ["init", "--upgrade"])
+        assert result.exit_code != 0
+
+    def test_upgrade_migrates_outdated_yml(self, cli_runner, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        (tmp_path / "tycoon.yml").write_text("name: old-project\nversion: 1.0.0\n")
+
+        result = cli_runner.invoke(app, ["init", "--upgrade"])
+
+        assert result.exit_code == 0
+        p = load_project(tmp_path)
+        assert p is not None
+        assert p.schema_version == SCHEMA_VERSION
+
+    def test_upgrade_already_current_reports_up_to_date(self, cli_runner, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        (tmp_path / "tycoon.yml").write_text(
+            f"name: current\nschema_version: {SCHEMA_VERSION}\n"
+            "metadata:\n  backend: duckdb_file\n  path: .tycoon/metadata.duckdb\n"
+        )
+
+        result = cli_runner.invoke(app, ["init", "--upgrade"])
+
+        assert result.exit_code == 0
+        assert "up to date" in result.stdout
+
+    def test_upgrade_future_schema_version_exits_nonzero_cleanly(self, cli_runner, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        (tmp_path / "tycoon.yml").write_text(f"name: future\nschema_version: {SCHEMA_VERSION + 1}\n")
+
+        result = cli_runner.invoke(app, ["init", "--upgrade"])
+
+        assert result.exit_code != 0
+        assert "newer than this tycoon supports" in result.output
diff --git a/tests/test_notify.py b/tests/test_notify.py
index 6cbc9f8..980d7ad 100644
--- a/tests/test_notify.py
+++ b/tests/test_notify.py
@@ -150,10 +150,7 @@ def _bind(self, tmp_path, monkeypatch):
             "sources: {}\n"
         )
         (tmp_path / "pyproject.toml").write_text('[project]\nname = "t"\n')
-        from tycoon.commands import run_all as ra_mod
-        from tycoon.config import TycoonConfig
-
-        monkeypatch.setattr(ra_mod, "config", TycoonConfig(project_root=tmp_path))
+        monkeypatch.chdir(tmp_path)
 
     def test_success_emits_notification(self, cli_runner, tmp_path, monkeypatch):
         self._bind(tmp_path, monkeypatch)
diff --git a/tests/test_project.py b/tests/test_project.py
index 45bb817..83d500d 100644
--- a/tests/test_project.py
+++ b/tests/test_project.py
@@ -2,7 +2,15 @@
 
 from __future__ import annotations
 
-from tycoon.project import DatabaseConfig, SourceConfig, TycoonProject, load_project, save_project
+from tycoon.project import (
+    DatabaseConfig,
+    SCHEMA_VERSION,
+    SourceConfig,
+    TycoonProject,
+    load_project,
+    migrate_project,
+    save_project,
+)
 
 
 class TestTycoonProject:
@@ -140,3 +148,162 @@ def test_config_sources_from_yml(self, tmp_path):
         )
         cfg = TycoonConfig(project_root=tmp_path)
         assert "my-api" in cfg.sources
+
+
+class TestRuntimesAndMetadata:
+    """T2-1: runtimes: and metadata: fields on TycoonProject."""
+
+    def test_existing_yml_loads_without_new_fields(self, tmp_path):
+        """A tycoon.yml with no runtimes/metadata keys must load with defaults."""
+        (tmp_path / "tycoon.yml").write_text("name: legacy-project\n")
+        p = load_project(tmp_path)
+        assert p is not None
+        assert p.runtimes == {}
+        assert p.metadata.backend == "duckdb_file"
+        assert p.metadata.path == ".tycoon/metadata.duckdb"
+
+    def test_runtimes_field_parses(self, tmp_path):
+        """runtimes: block with mixed types should parse into RuntimeEntry objects."""
+        (tmp_path / "tycoon.yml").write_text(
+            "name: runtimes-test\n"
+            "runtimes:\n"
+            "  shopify:\n"
+            "    type: dlt-managed\n"
+            "  custom_pipeline:\n"
+            "    type: dlt-project\n"
+            "    path: pipelines/custom\n"
+            "  fivetran_sync:\n"
+            "    type: fivetran\n"
+        )
+        p = load_project(tmp_path)
+        assert p is not None
+        assert set(p.runtimes) == {"shopify", "custom_pipeline", "fivetran_sync"}
+        assert p.runtimes["shopify"].type == "dlt-managed"
+        assert p.runtimes["shopify"].path is None
+        assert p.runtimes["custom_pipeline"].type == "dlt-project"
+        assert p.runtimes["custom_pipeline"].path == "pipelines/custom"
+        assert p.runtimes["fivetran_sync"].type == "fivetran"
+
+    def test_metadata_field_parses(self, tmp_path):
+        """metadata: block with custom values should override defaults."""
+        (tmp_path / "tycoon.yml").write_text(
+            "name: metadata-test\nmetadata:\n  backend: duckdb_file\n  path: .tycoon/custom_meta.duckdb\n"
+        )
+        p = load_project(tmp_path)
+        assert p is not None
+        assert p.metadata.backend == "duckdb_file"
+        assert p.metadata.path == ".tycoon/custom_meta.duckdb"
+
+
+class TestSchemaVersionEnforcement:
+    """T2-4: load_project is permissive; save_project preserves schema_version."""
+
+    def test_future_schema_version_loads_without_raise(self, tmp_path):
+        """load_project must not raise for a future schema_version — the gate
+        lives in load_config() so the import-time singleton never trips it."""
+        (tmp_path / "tycoon.yml").write_text(f"name: future\nschema_version: {SCHEMA_VERSION + 1}\n")
+        p = load_project(tmp_path)
+        assert p is not None
+        assert p.schema_version == SCHEMA_VERSION + 1
+
+    def test_save_project_preserves_schema_version(self, tmp_path):
+        """save_project preserves whatever schema_version is in the model;
+        only migrate_project (via init --upgrade) advances the stamp."""
+        (tmp_path / "tycoon.yml").write_text(f"name: old\nschema_version: {SCHEMA_VERSION}\n")
+        p = load_project(tmp_path)
+        assert p is not None
+        save_project(p, tmp_path)
+        reloaded = load_project(tmp_path)
+        assert reloaded is not None
+        assert reloaded.schema_version == SCHEMA_VERSION
+
+    def test_save_project_does_not_stamp_when_absent(self, tmp_path):
+        """A project with no schema_version on disk stays unstamped after save_project."""
+        (tmp_path / "tycoon.yml").write_text("name: old-project\n")
+        p = load_project(tmp_path)
+        assert p is not None
+        assert p.schema_version is None
+        save_project(p, tmp_path)
+        reloaded = load_project(tmp_path)
+        assert reloaded is not None
+        assert reloaded.schema_version is None
+
+
+class TestMigrateProject:
+    """T2-2: migrate_project writes missing keys and is idempotent."""
+
+    def test_missing_metadata_block_is_written(self, tmp_path):
+        """A yml without metadata: gets it added and schema_version stamped."""
+        (tmp_path / "tycoon.yml").write_text("name: old-project\nversion: 1.4.2\n")
+
+        modified = migrate_project(tmp_path)
+
+        assert modified is True
+        p = load_project(tmp_path)
+        assert p is not None
+        assert p.metadata.backend == "duckdb_file"
+        assert p.metadata.path == ".tycoon/metadata.duckdb"
+        assert p.schema_version == SCHEMA_VERSION
+
+    def test_user_version_not_overwritten(self, tmp_path):
+        """migrate_project never touches the user's version field."""
+        (tmp_path / "tycoon.yml").write_text("name: old-project\nversion: 1.4.2\n")
+
+        migrate_project(tmp_path)
+        p = load_project(tmp_path)
+
+        assert p is not None
+        assert p.version == "1.4.2"
+
+    def test_comments_preserved(self, tmp_path):
+        """Comments and blank lines survive the ruamel.yaml round-trip."""
+        original = "# Project config\nname: acme\n\n# owner: data-platform@acme.com\nversion: 1.0.0\n"
+        (tmp_path / "tycoon.yml").write_text(original)
+
+        migrate_project(tmp_path)
+        result = (tmp_path / "tycoon.yml").read_text()
+
+        assert "# Project config" in result
+        assert "# owner: data-platform@acme.com" in result
+
+    def test_second_call_is_no_op(self, tmp_path):
+        """Running migrate_project twice returns False on the second call."""
+        (tmp_path / "tycoon.yml").write_text("name: old-project\nversion: 0.1.0\n")
+
+        migrate_project(tmp_path)
+        modified_again = migrate_project(tmp_path)
+
+        assert modified_again is False
+
+    def test_already_migrated_yml_is_unchanged(self, tmp_path):
+        """A yml that already has metadata: and schema_version is left alone."""
+        (tmp_path / "tycoon.yml").write_text(
+            f"name: current-project\nschema_version: {SCHEMA_VERSION}\n"
+            "metadata:\n  backend: duckdb_file\n  path: .tycoon/metadata.duckdb\n"
+        )
+
+        modified = migrate_project(tmp_path)
+
+        assert modified is False
+
+    def test_future_schema_version_raises(self, tmp_path):
+        """A yml with schema_version newer than SCHEMA_VERSION raises ValueError."""
+        import pytest
+
+        (tmp_path / "tycoon.yml").write_text(f"name: future-project\nschema_version: {SCHEMA_VERSION + 1}\n")
+
+        with pytest.raises(ValueError, match="newer than this tycoon supports"):
+            migrate_project(tmp_path)
+
+    def test_non_integer_schema_version_raises(self, tmp_path):
+        """A float schema_version (e.g. 0.2 unquoted in YAML) raises ValueError."""
+        import pytest
+
+        (tmp_path / "tycoon.yml").write_text("name: bad-project\nschema_version: 0.2\n")
+
+        with pytest.raises(ValueError, match="must be an integer"):
+            migrate_project(tmp_path)
+
+    def test_missing_file_returns_false(self, tmp_path):
+        """migrate_project on a directory with no tycoon.yml returns False."""
+        assert migrate_project(tmp_path) is False
diff --git a/tests/test_sources.py b/tests/test_sources.py
index 055151a..05e6a94 100644
--- a/tests/test_sources.py
+++ b/tests/test_sources.py
@@ -67,10 +67,6 @@ class TestSourcesList:
     def test_list_shows_sources(self, cli_runner, tmp_path, monkeypatch):
         _setup_project(tmp_path)
         monkeypatch.chdir(tmp_path)
-        # Reload config for the new cwd
-        from tycoon.config import config
-
-        config.__init__(project_root=tmp_path)
 
         result = cli_runner.invoke(app, ["data", "sources", "list"])
         assert result.exit_code == 0
@@ -82,9 +78,6 @@ def test_list_shows_empty_message(self, cli_runner, tmp_path, monkeypatch):
         (tmp_path / "pyproject.toml").write_text('[project]\nname = "test"\n')
         (tmp_path / "tycoon.yml").write_text("name: empty\nsources: {}\n")
         monkeypatch.chdir(tmp_path)
-        from tycoon.config import config
-
-        config.__init__(project_root=tmp_path)
 
         result = cli_runner.invoke(app, ["data", "sources", "list"])
         assert result.exit_code == 0
@@ -92,9 +85,6 @@ def test_list_shows_empty_message(self, cli_runner, tmp_path, monkeypatch):
 
     def test_list_errors_without_project(self, cli_runner, tmp_path, monkeypatch):
         monkeypatch.chdir(tmp_path)
-        from tycoon.config import config
-
-        config.__init__(project_root=tmp_path)
 
         result = cli_runner.invoke(app, ["data", "sources", "list"])
         assert result.exit_code == 1
@@ -111,9 +101,6 @@ class TestSourcesShow:
     def test_show_existing_source(self, cli_runner, tmp_path, monkeypatch):
         _setup_project(tmp_path)
         monkeypatch.chdir(tmp_path)
-        from tycoon.config import config
-
-        config.__init__(project_root=tmp_path)
 
         # show is a subcommand of list: tycoon data sources list show <name>
         result = cli_runner.invoke(app, ["data", "sources", "list", "show", "nyc-dot"])
@@ -125,9 +112,6 @@ def test_show_existing_source(self, cli_runner, tmp_path, monkeypatch):
     def test_show_nonexistent_source(self, cli_runner, tmp_path, monkeypatch):
         _setup_project(tmp_path)
         monkeypatch.chdir(tmp_path)
-        from tycoon.config import config
-
-        config.__init__(project_root=tmp_path)
 
         result = cli_runner.invoke(app, ["data", "sources", "list", "show", "nonexistent"])
         assert result.exit_code == 1
@@ -145,9 +129,6 @@ class TestSourcesRemove:
     def test_remove_with_confirmation(self, cli_runner, tmp_path, monkeypatch):
         _setup_project(tmp_path)
         monkeypatch.chdir(tmp_path)
-        from tycoon.config import config
-
-        config.__init__(project_root=tmp_path)
 
         result = cli_runner.invoke(app, ["data", "sources", "remove", "nyc-dot"], input="y\n")
         assert result.exit_code == 0
@@ -161,9 +142,6 @@ def test_remove_with_confirmation(self, cli_runner, tmp_path, monkeypatch):
     def test_remove_abort(self, cli_runner, tmp_path, monkeypatch):
         _setup_project(tmp_path)
         monkeypatch.chdir(tmp_path)
-        from tycoon.config import config
-
-        config.__init__(project_root=tmp_path)
 
         result = cli_runner.invoke(app, ["data", "sources", "remove", "nyc-dot"], input="n\n")
         assert result.exit_code == 1  # typer.confirm abort exits 1
@@ -176,9 +154,6 @@ def test_remove_abort(self, cli_runner, tmp_path, monkeypatch):
     def test_remove_nonexistent_source(self, cli_runner, tmp_path, monkeypatch):
         _setup_project(tmp_path)
         monkeypatch.chdir(tmp_path)
-        from tycoon.config import config
-
-        config.__init__(project_root=tmp_path)
 
         result = cli_runner.invoke(app, ["data", "sources", "remove", "nonexistent"])
         assert result.exit_code == 1
@@ -328,22 +303,20 @@ def _seed_raw_db(self, path: Path, schema: str, table: str) -> None:
         finally:
             con.close()
 
-    def _bind_config(self, monkeypatch, project_root: Path) -> None:
-        from tycoon.commands import sources as sources_mod
+    def _make_cfg(self, project_root: Path):
         from tycoon.config import TycoonConfig
 
-        cfg = TycoonConfig(project_root=project_root)
-        monkeypatch.setattr(sources_mod, "config", cfg)
+        return TycoonConfig(project_root=project_root)
 
     def test_no_dbt_project_is_noop(self, tmp_path: Path, monkeypatch):
         from tycoon.commands.sources import _maybe_auto_scaffold
 
         _setup_project(tmp_path)
-        self._bind_config(monkeypatch, tmp_path)
+        cfg = self._make_cfg(tmp_path)
         # No dbt_project/ directory exists.
 
         sc = SourceConfig(type="rest_api", schema="raw_nyc_dot", config={})
-        _maybe_auto_scaffold("nyc-dot", sc, scaffold=True)
+        _maybe_auto_scaffold("nyc-dot", sc, cfg=cfg, scaffold=True)
         # No exception, no files written. Nothing to assert beyond "didn't raise."
 
     def test_skips_when_source_already_referenced(self, tmp_path: Path, monkeypatch):
@@ -355,10 +328,10 @@ def test_skips_when_source_already_referenced(self, tmp_path: Path, monkeypatch)
         models.mkdir(parents=True)
         (models / "stg_existing.sql").write_text("select * from {{ source('nyc-dot', 'i4gi-tjb9') }}\n")
         self._seed_raw_db(tmp_path / "data" / "raw.duckdb", "raw_nyc_dot", "i4gi_tjb9")
-        self._bind_config(monkeypatch, tmp_path)
+        cfg = self._make_cfg(tmp_path)
 
         sc = SourceConfig(type="rest_api", schema="raw_nyc_dot", config={})
-        _maybe_auto_scaffold("nyc-dot", sc, scaffold=True)
+        _maybe_auto_scaffold("nyc-dot", sc, cfg=cfg, scaffold=True)
 
         # No nyc-dot subdirectory was created.
         assert not (models / "nyc-dot").exists()
@@ -369,10 +342,10 @@ def test_generates_when_dbt_exists_and_no_prior_reference(self, tmp_path: Path,
         _setup_project(tmp_path)
         (tmp_path / "dbt_project" / "models").mkdir(parents=True)
         self._seed_raw_db(tmp_path / "data" / "raw.duckdb", "raw_nyc_dot", "i4gi_tjb9")
-        self._bind_config(monkeypatch, tmp_path)
+        cfg = self._make_cfg(tmp_path)
 
         sc = SourceConfig(type="rest_api", schema="raw_nyc_dot", config={})
-        _maybe_auto_scaffold("nyc-dot", sc, scaffold=True)
+        _maybe_auto_scaffold("nyc-dot", sc, cfg=cfg, scaffold=True)
 
         sql = tmp_path / "dbt_project" / "models" / "staging" / "nyc-dot" / "stg_nyc-dot__i4gi_tjb9.sql"
         assert sql.exists()
@@ -384,10 +357,10 @@ def test_no_scaffold_flag_skips(self, tmp_path: Path, monkeypatch):
         _setup_project(tmp_path)
         (tmp_path / "dbt_project" / "models").mkdir(parents=True)
         self._seed_raw_db(tmp_path / "data" / "raw.duckdb", "raw_nyc_dot", "i4gi_tjb9")
-        self._bind_config(monkeypatch, tmp_path)
+        cfg = self._make_cfg(tmp_path)
 
         sc = SourceConfig(type="rest_api", schema="raw_nyc_dot", config={})
-        _maybe_auto_scaffold("nyc-dot", sc, scaffold=False)
+        _maybe_auto_scaffold("nyc-dot", sc, cfg=cfg, scaffold=False)
 
         assert not (tmp_path / "dbt_project" / "models" / "staging" / "nyc-dot").exists()
 
@@ -399,10 +372,10 @@ def test_config_opt_out_skips(self, tmp_path: Path, monkeypatch):
         (tmp_path / "pyproject.toml").write_text('[project]\nname = "test"\n')
         (tmp_path / "dbt_project" / "models").mkdir(parents=True)
         self._seed_raw_db(tmp_path / "data" / "raw.duckdb", "raw_nyc_dot", "i4gi_tjb9")
-        self._bind_config(monkeypatch, tmp_path)
+        cfg = self._make_cfg(tmp_path)
 
         sc = SourceConfig(type="rest_api", schema="raw_nyc_dot", config={})
-        _maybe_auto_scaffold("nyc-dot", sc, scaffold=True)
+        _maybe_auto_scaffold("nyc-dot", sc, cfg=cfg, scaffold=True)
 
         assert not (tmp_path / "dbt_project" / "models" / "staging" / "nyc-dot").exists()
 
@@ -421,7 +394,7 @@ class TestSourcesAddNoPrompt:
     """
 
     def _bind(self, tmp_path: Path, monkeypatch):
-        """Set up an empty project + bind config to tmp_path."""
+        """Set up an empty project and chdir so commands find it."""
         body = (
             "name: test\n"
             "version: 0.1.0\n"
@@ -432,12 +405,7 @@ def _bind(self, tmp_path: Path, monkeypatch):
         )
         (tmp_path / "tycoon.yml").write_text(body)
         (tmp_path / "pyproject.toml").write_text('[project]\nname = "test"\n')
-        from tycoon.commands import sources as sources_mod
-        from tycoon.config import TycoonConfig
-
-        cfg = TycoonConfig(project_root=tmp_path)
-        monkeypatch.setattr(sources_mod, "config", cfg)
-        return cfg
+        monkeypatch.chdir(tmp_path)
 
     def test_rest_api_with_base_url_auto_derives_name_and_schema(self, cli_runner, tmp_path, monkeypatch):
         self._bind(tmp_path, monkeypatch)
diff --git a/tests/test_sync.py b/tests/test_sync.py
index 1380479..f63fa75 100644
--- a/tests/test_sync.py
+++ b/tests/test_sync.py
@@ -55,11 +55,6 @@ def project(tmp_path: Path, monkeypatch):
         "sources: {}\n"
     )
 
-    from tycoon.commands import sync_cmd as sync_mod
-    from tycoon.config import TycoonConfig
-
-    cfg = TycoonConfig(project_root=tmp_path)
-    monkeypatch.setattr(sync_mod, "config", cfg)
     monkeypatch.chdir(tmp_path)
     return tmp_path
 
@@ -339,12 +334,6 @@ def test_uses_tycoon_yml_sync_block(self, project, cli_runner, source_db):
             f"    - from: {source_db}\n"
             f"      schemas: ['mart']\n"
         )
-        # Re-rebind config to pick up the new tycoon.yml
-        from tycoon.commands import sync_cmd as sync_mod
-        from tycoon.config import TycoonConfig
-
-        sync_mod.config = TycoonConfig(project_root=project)
-
         result = cli_runner.invoke(app, ["data", "sync"])
         assert result.exit_code == 0, result.stdout
         assert dest.exists()
diff --git a/tests/test_templates_e2e.py b/tests/test_templates_e2e.py
index 190a70a..d353929 100644
--- a/tests/test_templates_e2e.py
+++ b/tests/test_templates_e2e.py
@@ -52,12 +52,10 @@ def _rebind_config(monkeypatch, project: Path) -> None:
     this patch they pick up the parent-tmpdir root pytest started in.
     """
     import tycoon.config as cfg_mod
-    from tycoon.commands import sources as sources_mod
     from tycoon.commands import transform as transform_mod
     from tycoon.config import TycoonConfig
 
     cfg = TycoonConfig(project_root=project)
-    monkeypatch.setattr(sources_mod, "config", cfg)
     monkeypatch.setattr(transform_mod, "config", cfg)
     monkeypatch.setattr(cfg_mod, "config", cfg)
 
@@ -322,12 +320,6 @@ def test_csv_import_then_data_sync(cli_runner, tmp_path, monkeypatch):
     _seed_widgets_csv(project / "data" / "input", count=5)
     _rebind_config(monkeypatch, project)
 
-    # Bind sync_cmd's config too, since it imports separately.
-    from tycoon.commands import sync_cmd as sync_mod
-    from tycoon.config import TycoonConfig
-
-    monkeypatch.setattr(sync_mod, "config", TycoonConfig(project_root=project))
-
     # ingest + transform
     assert cli_runner.invoke(app, ["data", "sources", "run", "files"]).exit_code == 0
     assert cli_runner.invoke(app, ["data", "transform", "run"]).exit_code == 0
diff --git a/uv.lock b/uv.lock
index b3ee75e..74f4279 100644
--- a/uv.lock
+++ b/uv.lock
@@ -225,6 +225,7 @@ dependencies = [
     { name = "pydantic" },
     { name = "pyyaml" },
     { name = "rich" },
+    { name = "ruamel-yaml" },
     { name = "typer" },
 ]
 
@@ -255,6 +256,7 @@ requires-dist = [
     { name = "pydantic", specifier = "==2.13.3" },
     { name = "pyyaml", specifier = "==6.0.3" },
     { name = "rich", specifier = "==15.0.0" },
+    { name = "ruamel-yaml", specifier = "==0.19.1" },
     { name = "typer", specifier = "==0.25.0" },
 ]
 provides-extras = ["docs"]
@@ -1507,6 +1509,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/0a/4d/222ef0b46443cf4cf46764d9c630f3fe4abaa7245be9417e56e9f52b8f65/rpds_py-0.30.0-cp313-cp313t-win_amd64.whl", hash = "sha256:f14fc5df50a716f7ece6a80b6c78bb35ea2ca47c499e422aa4463455dd96d56d", size = 225768 },
 ]
 
+[[package]]
+name = "ruamel-yaml"
+version = "0.19.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/c7/3b/ebda527b56beb90cb7652cb1c7e4f91f48649fbcd8d2eb2fb6e77cd3329b/ruamel_yaml-0.19.1.tar.gz", hash = "sha256:53eb66cd27849eff968ebf8f0bf61f46cdac2da1d1f3576dd4ccee9b25c31993", size = 142709 }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/b8/0c/51f6841f1d84f404f92463fc2b1ba0da357ca1e3db6b7fbda26956c3b82a/ruamel_yaml-0.19.1-py3-none-any.whl", hash = "sha256:27592957fedf6e0b62f281e96effd28043345e0e66001f97683aa9a40c667c93", size = 118102 },
+]
+
 [[package]]
 name = "ruff"
 version = "0.15.20"