diff --git a/amplifier_app_cli/main.py b/amplifier_app_cli/main.py
index e0e33e9..45ba958 100644
--- a/amplifier_app_cli/main.py
+++ b/amplifier_app_cli/main.py
@@ -2715,6 +2715,17 @@ def _extract_model_name() -> str:
                 )
         return "unknown"
 
+    # Helper to extract the active provider identity from config.
+    #
+    # Uses get_effective_config_summary() -- the same function the resume-time
+    # mismatch check in session_runner.py uses to compute the ACTIVE provider --
+    # so the value written here and the value compared at resume are always
+    # derived the same way. Returns the provider module id (e.g.
+    # "provider-anthropic"); session_runner._normalize_provider_identity()
+    # handles comparing this against older/bare-name forms.
+    def _extract_provider_identity() -> str:
+        return get_effective_config_summary(config, bundle_name).provider_module
+
     # Helper to save session after each turn
     async def _save_session():
         context = session.coordinator.get("context")
@@ -2731,6 +2742,7 @@ async def _save_session():
                 ),
                 "bundle": bundle_name,
                 "model": _extract_model_name(),
+                "provider": _extract_provider_identity(),
                 "turn_count": len([m for m in messages if m.get("role") == "user"]),
                 # Store working_dir for session sync between CLI and web
                 "working_dir": str(Path.cwd().resolve()),
@@ -3329,6 +3341,12 @@ async def execute_single(
                 ),
                 "bundle": bundle_name,
                 "model": model_name,
+                # Same source (get_effective_config_summary) as the resume-time
+                # mismatch check in session_runner.py -- see _extract_provider_identity()
+                # in interactive_chat() above for the parallel single-shot-vs-chat note.
+                "provider": get_effective_config_summary(
+                    config, bundle_name
+                ).provider_module,
                 "turn_count": len([m for m in messages if m.get("role") == "user"]),
                 # Store working_dir for session sync between CLI and web
                 "working_dir": str(Path.cwd().resolve()),
diff --git a/amplifier_app_cli/session_runner.py b/amplifier_app_cli/session_runner.py
index 16cb6d2..86e6041 100644
--- a/amplifier_app_cli/session_runner.py
+++ b/amplifier_app_cli/session_runner.py
@@ -15,6 +15,7 @@
 6. Register session spawning capability
 7. Restore transcript (resume only)
 7.5. Restore cumulative session cost (resume only)
+7.6. Warn (and confirm) on provider/model mismatch at resume (resume only)
 8. Register approval provider
 
 Philosophy:
@@ -25,19 +26,24 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import sys
 import uuid
 from collections import Counter
 from dataclasses import dataclass
 from dataclasses import field
+from datetime import UTC
+from datetime import datetime
 from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import Any
 
+import click
 from amplifier_core import AmplifierSession
 from amplifier_core import ModuleValidationError
 
+from .effective_config import get_effective_config_summary
 from .lib.settings import AppSettings
 from .session_store import SessionStore
 from .ui.error_display import display_validation_error
@@ -125,6 +131,7 @@ async def create_initialized_session(
     5. Register mention handling capability
     6. Register session spawning capability
     7. Restore transcript (resume only)
+    7.6. Warn (and confirm) on provider/model mismatch (resume only)
     8. Register approval provider
 
     Args:
@@ -290,6 +297,15 @@ async def create_initialized_session(
         except Exception:
             logger.debug("Prior session cost restore skipped", exc_info=True)
 
+    # Step 7.6: Warn (and, on a tty, confirm) on provider/model mismatch at
+    # resume - amplifier-support#208 Wave 2 / Option A. This is the ONE
+    # chokepoint every resume surface (amplifier run --resume, amplifier
+    # session resume, amplifier resume, amplifier continue) passes through,
+    # before the first session.execute() call. Runs only for resumes; silent
+    # (no console output, no prompt) when there's nothing to warn about.
+    if config.is_resume:
+        await _warn_on_resume_provider_mismatch(config, session_id, console, session)
+
     # Step 10: Register approval provider (app-layer policy)
     from .approval_provider import CLIApprovalProvider
     from .stdin_arbiter import StdinArbiter
@@ -333,6 +349,215 @@ async def create_initialized_session(
     )
 
 
+# =============================================================================
+# Resume-time provider/model mismatch check (amplifier-support#208, Wave 2)
+# =============================================================================
+#
+# Sessions that switch providers mid-conversation can persist provider-specific
+# content (e.g. Anthropic thinking-block signatures) that bricks the session
+# when replayed against a different provider (400s on invalid signatures).
+# Wave 1 sanitized/repaired that content at the provider boundary. This is
+# the resume-time guardrail (Option A, ack'd by the maintainer): warn -- and,
+# interactively, require confirmation -- when the provider/model about to
+# handle a resumed session differs from whichever provider/model last wrote
+# its metadata.
+
+
+def _normalize_provider_identity(value: str | None) -> str:
+    """Normalize a provider identity for mismatch comparison.
+
+    Provider identity may be recorded as a module id ("provider-anthropic")
+    or a bare provider name ("anthropic") depending on the source/vintage of
+    the metadata. Strip the "provider-" prefix and lowercase so both forms
+    compare equal, regardless of which side (stored metadata vs. active
+    config) uses which form.
+
+    Mirrors the nested _normalize_to_provider_name() helper inside
+    _should_attempt_self_healing() above, but is exposed at module level
+    since it is also needed here to compare against the "provider" field
+    persisted by main.py's session-save code.
+
+    Args:
+        value: Provider identity string (module id or bare name), or None.
+
+    Returns:
+        Lowercased bare provider name (e.g. "anthropic"), or "" if value is
+        falsy.
+    """
+    if not value:
+        return ""
+    normalized = value.strip().lower()
+    if normalized.startswith("provider-"):
+        normalized = normalized[len("provider-") :]
+    return normalized
+
+
+async def _warn_on_resume_provider_mismatch(
+    config: SessionConfig,
+    session_id: str,
+    console: "Console",
+    session: AmplifierSession,
+) -> None:
+    """Warn (and, on a tty, require confirmation) on provider/model mismatch.
+
+    Compares the provider/model that is about to handle this resumed session
+    (the ACTIVE config, resolved with any --provider/--model overrides already
+    applied) against whichever provider/model last wrote this session's
+    metadata.json (the PRIOR writer).
+
+    Feature-detection: pre-existing sessions (saved before this PR) have no
+    "provider" field in their metadata. For those, comparison falls back to
+    model-only. Sessions saved by this PR (or later) carry "provider" and get
+    the full provider+model comparison.
+
+    On match, or when there is no prior metadata to compare against at all:
+    completely silent -- zero behavior change and zero console output.
+
+    On mismatch:
+    - stdin is a tty: print the warning, then require explicit confirmation
+      via `click.confirm` (offloaded to a worker thread -- see the
+      KeyboardInterrupt handler in main.py for why this offload is mandatory:
+      click.confirm() performs a blocking, synchronous stdin read that would
+      otherwise freeze the event loop). Declining raises click.Abort(),
+      which aborts the resume before any session.execute() can run.
+    - stdin is not a tty (CI, piped input, scripted flows): print the warning
+      and continue automatically. Scripted flows must never hang waiting for
+      input that will never arrive.
+
+    This function is best-effort with respect to loading prior metadata: any
+    failure there (corrupt/missing metadata.json, invalid session_id, etc.)
+    is swallowed and treated as "nothing to compare against" -- this check
+    must never turn a resume that would otherwise have succeeded into a
+    failure.
+
+    Args:
+        config: The (already resume-mode) SessionConfig for this session.
+            config.config is the fully-resolved active configuration (CLI
+            --provider/--model overrides are applied before this point, so
+            comparing against it automatically respects them).
+        session_id: The resolved session id being resumed.
+        console: Rich console for the warning. In JSON output modes,
+            create_initialized_session's caller has already redirected
+            console.file to stderr before create_initialized_session is
+            invoked, so this can never contaminate JSON stdout.
+        session: The already-created AmplifierSession (Steps 1-7.5 have run
+            by this point). On decline, this is cleaned up before raising
+            click.Abort so we never leave a half-initialized session behind.
+
+    Raises:
+        click.Abort: if stdin is a tty and the user declines to continue.
+    """
+    try:
+        prior_metadata = SessionStore().get_metadata(session_id)
+    except Exception:
+        logger.debug(
+            "Resume provider-mismatch check: could not load prior metadata "
+            "for session %s",
+            session_id,
+            exc_info=True,
+        )
+        return
+
+    prior_model = prior_metadata.get("model")
+    prior_provider = prior_metadata.get("provider")
+
+    # Nothing recorded to compare against (e.g. corrupted/minimal recovered
+    # metadata, or a session saved before "model" was ever written) -- there
+    # is nothing to warn about.
+    if not prior_model and not prior_provider:
+        return
+
+    summary = get_effective_config_summary(config.config, config.bundle_name)
+    active_provider = summary.provider_module
+    active_model = summary.model
+
+    if prior_provider:
+        # Full comparison: normalized provider identity AND model.
+        if (
+            _normalize_provider_identity(prior_provider)
+            == _normalize_provider_identity(active_provider)
+            and prior_model == active_model
+        ):
+            return  # Exact match -- silent, zero behavior change.
+    else:
+        # Feature-detection fallback for pre-existing sessions that predate
+        # the "provider" metadata field: compare model only.
+        if prior_model == active_model:
+            return  # Silent -- nothing more to check without a prior provider.
+
+    prior_label = f"{prior_provider or 'unknown'}/{prior_model or 'unknown'}"
+    active_label = f"{active_provider}/{active_model}"
+
+    console.print(
+        f"[yellow]\u26a0 Provider/model mismatch:[/yellow] session was last "
+        f"written by {prior_label} \u2014 now resuming with {active_label}\n"
+        "[dim]  Cross-provider replay can fail on provider-specific content "
+        "(e.g. thinking blocks).[/dim]"
+    )
+
+    if sys.stdin.isatty():
+        # click.confirm() performs a synchronous, canonical-mode blocking
+        # stdin read (input()) with no executor offload. Calling it directly
+        # here would block the ENTIRE asyncio event loop thread until Enter
+        # is pressed. Offload to a worker thread, mirroring the existing
+        # pattern in main.py's KeyboardInterrupt handler and
+        # approval_provider.py's _get_user_input().
+        if not await asyncio.to_thread(
+            click.confirm,
+            "Continue resuming with the new provider?",
+            default=False,
+        ):
+            console.print("[dim]Resume cancelled.[/dim]")
+            # Never leave a half-initialized session behind: Steps 1-7.5 have
+            # already created and partially wired up `session` by this point.
+            try:
+                await session.cleanup()
+            except Exception:
+                logger.debug(
+                    "Session cleanup after declined resume failed", exc_info=True
+                )
+            raise click.Abort()
+        _record_provider_mismatch(session_id, prior_label, active_label)
+    else:
+        # Non-interactive (CI, piped stdin, shadow environments): warn-only.
+        # Scripted flows must not hang waiting for input that can't arrive.
+        _record_provider_mismatch(session_id, prior_label, active_label)
+
+
+def _record_provider_mismatch(session_id: str, prior: str, active: str) -> None:
+    """Best-effort audit trail for an accepted (or warned-through) mismatch.
+
+    Mirrors _record_bundle_override() in commands/session.py: appends a
+    timestamped entry to a metadata list rather than overwriting anything.
+    Failures here are swallowed -- this is a diagnostics nicety, never a
+    condition that should fail the resume itself.
+
+    Args:
+        session_id: The session whose metadata should record the mismatch.
+        prior: Display label for the provider/model that last wrote the
+            session (e.g. "provider-anthropic/claude-x").
+        active: Display label for the provider/model now resuming it.
+    """
+    try:
+        store = SessionStore()
+        metadata = store.get_metadata(session_id)
+        mismatches = list(metadata.get("provider_mismatches", []))
+        mismatches.append(
+            {
+                "timestamp": datetime.now(UTC).isoformat(timespec="milliseconds"),
+                "previous": prior,
+                "resumed_with": active,
+            }
+        )
+        store.update_metadata(session_id, {"provider_mismatches": mismatches})
+    except Exception:
+        logger.debug(
+            "Could not record provider-mismatch audit trail for session %s",
+            session_id,
+            exc_info=True,
+        )
+
+
 _CLEANUP_EVENTS: tuple[str, ...] = (
     # PR #183 — cleanup-window diagnostic events emitted by app-cli's main.py only
     "cleanup:render_begin",
diff --git a/tests/test_resume_provider_metadata_write.py b/tests/test_resume_provider_metadata_write.py
new file mode 100644
index 0000000..d1af5be
--- /dev/null
+++ b/tests/test_resume_provider_metadata_write.py
@@ -0,0 +1,173 @@
+"""Tests for write-side provider persistence in session metadata.
+
+Wave 2 of the cross-provider resume hardening set (amplifier-support#208):
+this repo's metadata-writing code -- both the interactive `_save_session()`
+closure inside `interactive_chat()` and the single-shot save block inside
+`execute_single()` -- must persist the active `"provider"` identity
+alongside the existing `"model"` field. Without this, the resume-time
+mismatch check (session_runner._warn_on_resume_provider_mismatch) has
+nothing new to compare against for sessions saved going forward.
+
+Both write sites derive `"provider"` from `get_effective_config_summary()`
+(`.provider_module`) -- the same function the read-side check uses to
+compute the ACTIVE provider at resume -- so write-side and read-side always
+agree on what "provider identity" means and how it's shaped (module id form,
+e.g. "provider-anthropic").
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+_MODULE = "amplifier_app_cli.main"
+
+
+def _make_mock_hooks() -> MagicMock:
+    """Return a mock hooks registry whose emit() is an inert AsyncMock."""
+    mock = MagicMock()
+    mock.emit = AsyncMock(return_value=None)
+    return mock
+
+
+def _make_mock_session(hooks: MagicMock, providers: dict | None = None) -> MagicMock:
+    """Return a minimal mock AmplifierSession suitable for execute_single()."""
+    mock_ctx = MagicMock()
+    mock_ctx.get_messages = AsyncMock(return_value=[{"role": "user", "content": "Hi"}])
+
+    def _coordinator_get(key: str):
+        if key == "hooks":
+            return hooks
+        if key == "context":
+            return mock_ctx
+        if key == "providers":
+            return providers if providers is not None else {}
+        return None
+
+    mock_session = MagicMock()
+    mock_session.session_id = "test-session-id"
+    mock_session.execute = AsyncMock(return_value="Hello!")
+    mock_session.coordinator = MagicMock()
+    mock_session.coordinator.get = _coordinator_get
+    mock_session.coordinator.get_capability.return_value = None
+    mock_session.coordinator.cancellation = MagicMock()
+    mock_session.coordinator.cancellation.is_cancelled = False
+    return mock_session
+
+
+def _make_mock_initialized(session: MagicMock) -> MagicMock:
+    """Return a minimal InitializedSession mock wrapping *session*."""
+    mock = MagicMock()
+    mock.session = session
+    mock.session_id = "test-session-id"
+    mock.cleanup = AsyncMock()
+    return mock
+
+
+class TestExecuteSingleWritesProvider:
+    """execute_single()'s final metadata-save block persists 'provider'."""
+
+    @pytest.mark.asyncio
+    async def test_metadata_includes_provider_matching_config(self, tmp_path: Path):
+        """The saved metadata's 'provider' matches the resolved config's provider_module."""
+        from amplifier_app_cli.main import execute_single
+
+        hooks = _make_mock_hooks()
+        session = _make_mock_session(hooks)
+        initialized = _make_mock_initialized(session)
+
+        config = {
+            "providers": [
+                {
+                    "module": "provider-anthropic",
+                    "config": {"default_model": "claude-x", "priority": 0},
+                }
+            ]
+        }
+
+        with (
+            patch(
+                f"{_MODULE}.create_initialized_session",
+                new=AsyncMock(return_value=initialized),
+            ),
+            patch(f"{_MODULE}.SessionStore") as MockStore,
+            patch(f"{_MODULE}.console"),
+            patch(
+                f"{_MODULE}.process_runtime_mentions",
+                new=AsyncMock(side_effect=lambda _s, p: p),
+            ),
+        ):
+            store_instance = MockStore.return_value
+            store_instance.get_metadata.return_value = {}
+            store_instance.save.return_value = None
+
+            await execute_single(
+                prompt="Hi",
+                config=config,
+                search_paths=[tmp_path],
+                verbose=False,
+                output_format="text",
+                bundle_name="bundle:test",
+            )
+
+        store_instance.save.assert_called_once()
+        _saved_id, _saved_messages, saved_metadata = store_instance.save.call_args[0]
+        assert saved_metadata.get("provider") == "provider-anthropic"
+        # Sanity: 'model' is still written too (pre-existing field, untouched by this PR).
+        assert "model" in saved_metadata
+
+    @pytest.mark.asyncio
+    async def test_metadata_preserves_existing_fields_alongside_provider(
+        self, tmp_path: Path
+    ):
+        """Existing metadata fields (name, description, ...) survive the save,
+        with 'provider' added alongside them -- not replacing them."""
+        from amplifier_app_cli.main import execute_single
+
+        hooks = _make_mock_hooks()
+        session = _make_mock_session(hooks)
+        initialized = _make_mock_initialized(session)
+
+        config = {
+            "providers": [
+                {
+                    "module": "provider-openai",
+                    "config": {"default_model": "gpt-x", "priority": 0},
+                }
+            ]
+        }
+
+        with (
+            patch(
+                f"{_MODULE}.create_initialized_session",
+                new=AsyncMock(return_value=initialized),
+            ),
+            patch(f"{_MODULE}.SessionStore") as MockStore,
+            patch(f"{_MODULE}.console"),
+            patch(
+                f"{_MODULE}.process_runtime_mentions",
+                new=AsyncMock(side_effect=lambda _s, p: p),
+            ),
+        ):
+            store_instance = MockStore.return_value
+            store_instance.get_metadata.return_value = {
+                "name": "my-named-session",
+                "description": "a test session",
+            }
+            store_instance.save.return_value = None
+
+            await execute_single(
+                prompt="Hi",
+                config=config,
+                search_paths=[tmp_path],
+                verbose=False,
+                output_format="text",
+                bundle_name="bundle:test",
+            )
+
+        _saved_id, _saved_messages, saved_metadata = store_instance.save.call_args[0]
+        assert saved_metadata["provider"] == "provider-openai"
+        assert saved_metadata["name"] == "my-named-session"
+        assert saved_metadata["description"] == "a test session"
diff --git a/tests/test_session_runner.py b/tests/test_session_runner.py
index b7b528c..c3b945c 100644
--- a/tests/test_session_runner.py
+++ b/tests/test_session_runner.py
@@ -575,9 +575,7 @@ def _configurator_patches(mock_sess):
             new_callable=AsyncMock,
             return_value=mock_sess,
         ),
-        patch(
-            "amplifier_app_cli.commands.init.check_first_run", return_value=False
-        ),
+        patch("amplifier_app_cli.commands.init.check_first_run", return_value=False),
         patch(
             "amplifier_app_cli.project_utils.get_project_slug",
             return_value="test-slug",
@@ -852,3 +850,348 @@ async def test_configurator_none_on_general_exception(self, caplog):
             r.message for r in caplog.records if r.levelno == logging.WARNING
         ]
         assert warning_messages, "Expected a warning but none was logged"
+
+
+# ---------------------------------------------------------------------------
+# Resume-time provider/model mismatch check (amplifier-support#208, Wave 2)
+# ---------------------------------------------------------------------------
+
+import click as _click  # noqa: E402  (grouped here to stay close to its tests)
+
+from amplifier_app_cli.effective_config import EffectiveConfigSummary  # noqa: E402
+from amplifier_app_cli.session_runner import (  # noqa: E402
+    _normalize_provider_identity,
+    _warn_on_resume_provider_mismatch,
+)
+
+
+def _make_summary(
+    provider_module: str = "provider-anthropic", model: str = "claude-x"
+) -> EffectiveConfigSummary:
+    """Return a minimal EffectiveConfigSummary for mismatch-check tests."""
+    return EffectiveConfigSummary(
+        config_source="bundle:test",
+        provider_name=provider_module.replace("provider-", "").title(),
+        provider_module=provider_module,
+        model=model,
+        orchestrator="loop-basic",
+        tool_count=0,
+        hook_count=0,
+    )
+
+
+class TestNormalizeProviderIdentity:
+    """Unit tests for _normalize_provider_identity()."""
+
+    def test_strips_provider_prefix(self):
+        assert _normalize_provider_identity("provider-anthropic") == "anthropic"
+
+    def test_bare_name_passes_through_unchanged(self):
+        assert _normalize_provider_identity("anthropic") == "anthropic"
+
+    def test_case_insensitive(self):
+        assert _normalize_provider_identity("Provider-Anthropic") == "anthropic"
+        assert _normalize_provider_identity("ANTHROPIC") == "anthropic"
+
+    def test_none_and_empty_string_normalize_to_empty(self):
+        assert _normalize_provider_identity(None) == ""
+        assert _normalize_provider_identity("") == ""
+
+
+class TestWarnOnResumeProviderMismatch:
+    """Unit tests for _warn_on_resume_provider_mismatch() -- the read-side policy."""
+
+    @staticmethod
+    def _resume_config() -> SessionConfig:
+        return _make_session_config(
+            initial_transcript=[{"role": "user", "content": "hi"}]
+        )
+
+    @pytest.mark.asyncio
+    async def test_mismatch_tty_prints_warning_and_confirms_then_proceeds(self):
+        """mismatch + tty -> warning printed AND confirm invoked; accept proceeds."""
+        cfg = self._resume_config()
+        console = MagicMock()
+        session = MagicMock()
+        session.cleanup = AsyncMock()
+
+        with (
+            patch(f"{_MODULE}.SessionStore") as MockStore,
+            patch(
+                f"{_MODULE}.get_effective_config_summary",
+                return_value=_make_summary(
+                    provider_module="provider-openai", model="gpt-x"
+                ),
+            ),
+            patch("sys.stdin.isatty", return_value=True),
+            patch(f"{_MODULE}.click.confirm", return_value=True) as mock_confirm,
+        ):
+            MockStore.return_value.get_metadata.return_value = {
+                "model": "claude-x",
+                "provider": "provider-anthropic",
+            }
+            await _warn_on_resume_provider_mismatch(cfg, "sess-1", console, session)
+
+        console.print.assert_called()
+        mock_confirm.assert_called_once()
+        session.cleanup.assert_not_called()
+        printed = " ".join(str(c) for c in console.print.call_args_list)
+        assert "provider-anthropic/claude-x" in printed
+        assert "provider-openai/gpt-x" in printed
+
+    @pytest.mark.asyncio
+    async def test_mismatch_tty_decline_aborts_cleanly(self):
+        """mismatch + tty + decline -> click.Abort raised, session cleaned up."""
+        cfg = self._resume_config()
+        console = MagicMock()
+        session = MagicMock()
+        session.cleanup = AsyncMock()
+
+        with (
+            patch(f"{_MODULE}.SessionStore") as MockStore,
+            patch(
+                f"{_MODULE}.get_effective_config_summary",
+                return_value=_make_summary(
+                    provider_module="provider-openai", model="gpt-x"
+                ),
+            ),
+            patch("sys.stdin.isatty", return_value=True),
+            patch(f"{_MODULE}.click.confirm", return_value=False),
+        ):
+            MockStore.return_value.get_metadata.return_value = {
+                "model": "claude-x",
+                "provider": "provider-anthropic",
+            }
+            with pytest.raises(_click.Abort):
+                await _warn_on_resume_provider_mismatch(cfg, "sess-1", console, session)
+
+        session.cleanup.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_mismatch_non_tty_warns_and_continues_without_confirm(self):
+        """mismatch + non-tty -> warning printed, NO confirm, proceeds automatically."""
+        cfg = self._resume_config()
+        console = MagicMock()
+        session = MagicMock()
+        session.cleanup = AsyncMock()
+
+        with (
+            patch(f"{_MODULE}.SessionStore") as MockStore,
+            patch(
+                f"{_MODULE}.get_effective_config_summary",
+                return_value=_make_summary(
+                    provider_module="provider-openai", model="gpt-x"
+                ),
+            ),
+            patch("sys.stdin.isatty", return_value=False),
+            patch(f"{_MODULE}.click.confirm") as mock_confirm,
+        ):
+            MockStore.return_value.get_metadata.return_value = {
+                "model": "claude-x",
+                "provider": "provider-anthropic",
+            }
+            await _warn_on_resume_provider_mismatch(cfg, "sess-1", console, session)
+
+        console.print.assert_called()
+        mock_confirm.assert_not_called()
+        session.cleanup.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_provider_and_model_match_is_completely_silent(self):
+        """Exact provider+model match -> no warning, no confirm (zero behavior change)."""
+        cfg = self._resume_config()
+        console = MagicMock()
+        session = MagicMock()
+
+        with (
+            patch(f"{_MODULE}.SessionStore") as MockStore,
+            patch(
+                f"{_MODULE}.get_effective_config_summary",
+                return_value=_make_summary(
+                    provider_module="provider-anthropic", model="claude-x"
+                ),
+            ),
+            patch("sys.stdin.isatty", return_value=True),
+            patch(f"{_MODULE}.click.confirm") as mock_confirm,
+        ):
+            MockStore.return_value.get_metadata.return_value = {
+                "model": "claude-x",
+                "provider": "provider-anthropic",
+            }
+            await _warn_on_resume_provider_mismatch(cfg, "sess-1", console, session)
+
+        console.print.assert_not_called()
+        mock_confirm.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_missing_provider_field_model_differs_falls_back_to_model_only(self):
+        """Pre-existing session with no 'provider' field: model-only fallback still warns."""
+        cfg = self._resume_config()
+        console = MagicMock()
+        session = MagicMock()
+
+        with (
+            patch(f"{_MODULE}.SessionStore") as MockStore,
+            patch(
+                f"{_MODULE}.get_effective_config_summary",
+                return_value=_make_summary(
+                    provider_module="provider-openai", model="gpt-x"
+                ),
+            ),
+            patch("sys.stdin.isatty", return_value=False),
+            patch(f"{_MODULE}.click.confirm") as mock_confirm,
+        ):
+            # No "provider" key at all -- simulates a session saved before this PR.
+            MockStore.return_value.get_metadata.return_value = {"model": "claude-x"}
+            await _warn_on_resume_provider_mismatch(cfg, "sess-1", console, session)
+
+        console.print.assert_called()
+        mock_confirm.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_missing_provider_field_model_matches_is_silent(self):
+        """Pre-existing session, no 'provider' field, model matches -> silent."""
+        cfg = self._resume_config()
+        console = MagicMock()
+        session = MagicMock()
+
+        with (
+            patch(f"{_MODULE}.SessionStore") as MockStore,
+            patch(
+                f"{_MODULE}.get_effective_config_summary",
+                return_value=_make_summary(
+                    provider_module="provider-anthropic", model="claude-x"
+                ),
+            ),
+            patch("sys.stdin.isatty", return_value=True),
+            patch(f"{_MODULE}.click.confirm") as mock_confirm,
+        ):
+            MockStore.return_value.get_metadata.return_value = {"model": "claude-x"}
+            await _warn_on_resume_provider_mismatch(cfg, "sess-1", console, session)
+
+        console.print.assert_not_called()
+        mock_confirm.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_provider_normalization_prevents_false_positive(self):
+        """Stored 'provider-anthropic' vs active bare 'anthropic' -> normalized, no warning."""
+        cfg = self._resume_config()
+        console = MagicMock()
+        session = MagicMock()
+
+        with (
+            patch(f"{_MODULE}.SessionStore") as MockStore,
+            patch(
+                f"{_MODULE}.get_effective_config_summary",
+                return_value=_make_summary(
+                    provider_module="anthropic", model="claude-x"
+                ),
+            ),
+            patch("sys.stdin.isatty", return_value=True),
+            patch(f"{_MODULE}.click.confirm") as mock_confirm,
+        ):
+            MockStore.return_value.get_metadata.return_value = {
+                "model": "claude-x",
+                "provider": "provider-anthropic",
+            }
+            await _warn_on_resume_provider_mismatch(cfg, "sess-1", console, session)
+
+        console.print.assert_not_called()
+        mock_confirm.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_no_prior_metadata_at_all_is_silent(self):
+        """Empty prior metadata (nothing to compare against) -> silent."""
+        cfg = self._resume_config()
+        console = MagicMock()
+        session = MagicMock()
+
+        with (
+            patch(f"{_MODULE}.SessionStore") as MockStore,
+            patch(
+                f"{_MODULE}.get_effective_config_summary", return_value=_make_summary()
+            ),
+            patch(f"{_MODULE}.click.confirm") as mock_confirm,
+        ):
+            MockStore.return_value.get_metadata.return_value = {}
+            await _warn_on_resume_provider_mismatch(cfg, "sess-1", console, session)
+
+        console.print.assert_not_called()
+        mock_confirm.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_prior_metadata_load_failure_is_silent_best_effort(self):
+        """Loading prior metadata fails (e.g. corrupted session) -> silent no-op."""
+        cfg = self._resume_config()
+        console = MagicMock()
+        session = MagicMock()
+
+        with (
+            patch(f"{_MODULE}.SessionStore") as MockStore,
+            patch(
+                f"{_MODULE}.get_effective_config_summary", return_value=_make_summary()
+            ),
+        ):
+            MockStore.return_value.get_metadata.side_effect = FileNotFoundError("nope")
+            await _warn_on_resume_provider_mismatch(cfg, "sess-1", console, session)
+
+        console.print.assert_not_called()
+
+
+class TestProviderMismatchCheckWiredIntoChokepoint:
+    """Integration: create_initialized_session gates the check on config.is_resume.
+
+    This is the single chokepoint every resume surface passes through
+    (amplifier run --resume, amplifier session resume, amplifier resume,
+    amplifier continue all funnel through create_initialized_session).
+    """
+
+    @pytest.mark.asyncio
+    async def test_check_never_runs_for_a_new_non_resume_session(self):
+        """is_resume=False (new session) -> the mismatch check must never run."""
+        from contextlib import ExitStack
+
+        mock_sess = _make_mock_session()
+        cfg = _make_session_config()  # no initial_transcript -> is_resume is False
+        console = MagicMock()
+
+        with ExitStack() as stack:
+            for p in _configurator_patches(mock_sess):
+                stack.enter_context(p)
+            mock_check = stack.enter_context(
+                patch(
+                    f"{_MODULE}._warn_on_resume_provider_mismatch",
+                    new_callable=AsyncMock,
+                )
+            )
+            await create_initialized_session(cfg, console)
+
+        mock_check.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_check_runs_for_a_resume_session_with_correct_args(self):
+        """is_resume=True (resume) -> the mismatch check runs with (config, session_id, console, session)."""
+        from contextlib import ExitStack
+
+        mock_sess = _make_mock_session()
+        cfg = _make_session_config(
+            initial_transcript=[{"role": "user", "content": "hi"}]
+        )
+        console = MagicMock()
+
+        with ExitStack() as stack:
+            for p in _configurator_patches(mock_sess):
+                stack.enter_context(p)
+            mock_check = stack.enter_context(
+                patch(
+                    f"{_MODULE}._warn_on_resume_provider_mismatch",
+                    new_callable=AsyncMock,
+                )
+            )
+            result = await create_initialized_session(cfg, console)
+
+        mock_check.assert_called_once()
+        call_args = mock_check.call_args[0]
+        assert call_args[0] is cfg
+        assert call_args[2] is console
+        assert call_args[3] is result.session