diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1e35f6af..e11046eb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Diff-aware `apm install` — manifest as source of truth: removed packages, ref/version changes, and MCP config drift in `apm.yml` all self-correct on the next `apm install` without `--update` or `--force`; introduces `drift.py` with pure helper functions (#260)
+
 ## [0.7.7] - 2026-03-10
 
 ### Added
diff --git a/docs/src/content/docs/reference/cli-commands.md b/docs/src/content/docs/reference/cli-commands.md
index f4e84930..168b4a90 100644
--- a/docs/src/content/docs/reference/cli-commands.md
+++ b/docs/src/content/docs/reference/cli-commands.md
@@ -92,6 +92,13 @@ apm install [PACKAGES...] [OPTIONS]
 - `apm install` (no args): Installs **all** packages from `apm.yml`
 - `apm install <package>`: Installs **only** the specified package (adds to `apm.yml` if not present)
 
+**Diff-Aware Installation (manifest as source of truth):**
+- MCP servers already configured with matching config are skipped (`already configured`)
+- MCP servers already configured but with changed manifest config are re-applied automatically (`updated`)
+- APM packages removed from `apm.yml` have their deployed files cleaned up on the next full `apm install`
+- APM packages whose ref/version changed in `apm.yml` are re-downloaded automatically (no `--update` needed)
+- `--force` remains available for full overwrite/reset scenarios
+
 **Examples:**
 ```bash
 # Install all dependencies from apm.yml
diff --git a/docs/src/content/docs/reference/lockfile-spec.md b/docs/src/content/docs/reference/lockfile-spec.md
index 36f3559d..2d007f29 100644
--- a/docs/src/content/docs/reference/lockfile-spec.md
+++ b/docs/src/content/docs/reference/lockfile-spec.md
@@ -101,6 +101,7 @@ dependencies:
 | `apm_version` | string | MUST | Version of APM that generated this lock file. |
 | `dependencies` | array | MUST | Ordered list of resolved dependencies (see [section 4.2](#42-dependency-entries)). |
 | `mcp_servers` | array | MAY | List of MCP server identifiers registered by installed packages. |
+| `mcp_configs` | mapping | MAY | Mapping of MCP server name to its manifest configuration dict. Used for diff-aware installation — when config in `apm.yml` changes, `apm install` detects the drift and re-applies without `--force`. |
 
 ### 4.2 Dependency Entries
 
@@ -264,6 +265,11 @@ dependencies:
 
 mcp_servers:
   - security-scanner
+
+mcp_configs:
+  security-scanner:
+    name: security-scanner
+    transport: stdio
 ```
 
 ---
diff --git a/src/apm_cli/commands/install.py b/src/apm_cli/commands/install.py
index 3d1f069a..37447d93 100644
--- a/src/apm_cli/commands/install.py
+++ b/src/apm_cli/commands/install.py
@@ -7,6 +7,7 @@
 
 import click
 
+from ..drift import build_download_ref, detect_orphans, detect_ref_change
 from ..utils.console import _rich_error, _rich_info, _rich_success, _rich_warning
 from ..utils.diagnostics import DiagnosticCollector
 from ..utils.github_host import default_host, is_valid_fqdn
@@ -391,15 +392,17 @@ def install(ctx, packages, runtime, exclude, only, update, dry_run, force, verbo
         prompt_count = 0
         agent_count = 0
 
-        # Capture old MCP servers from lockfile BEFORE _install_apm_dependencies
-        # regenerates it (which drops the mcp_servers field).
+        # Capture old MCP servers and configs from lockfile BEFORE
+        # _install_apm_dependencies regenerates it (which drops the fields).
         # We always read this — even when --only=apm — so we can restore the
         # field after the lockfile is regenerated by the APM install step.
         old_mcp_servers: builtins.set = builtins.set()
+        old_mcp_configs: builtins.dict = {}
         _lock_path = Path.cwd() / "apm.lock"
         _existing_lock = LockFile.read(_lock_path)
         if _existing_lock:
             old_mcp_servers = builtins.set(_existing_lock.mcp_servers)
+            old_mcp_configs = builtins.dict(_existing_lock.mcp_configs)
 
         apm_diagnostics = None
         if should_install_apm and apm_deps:
@@ -441,26 +444,30 @@ def install(ctx, packages, runtime, exclude, only, update, dry_run, force, verbo
         mcp_count = 0
         new_mcp_servers: builtins.set = builtins.set()
         if should_install_mcp and mcp_deps:
-            mcp_count = MCPIntegrator.install(mcp_deps, runtime, exclude, verbose)
+            mcp_count = MCPIntegrator.install(
+                mcp_deps, runtime, exclude, verbose,
+                stored_mcp_configs=old_mcp_configs,
+            )
             new_mcp_servers = MCPIntegrator.get_server_names(mcp_deps)
+            new_mcp_configs = MCPIntegrator.get_server_configs(mcp_deps)
 
             # Remove stale MCP servers that are no longer needed
             stale_servers = old_mcp_servers - new_mcp_servers
             if stale_servers:
                 MCPIntegrator.remove_stale(stale_servers, runtime, exclude)
 
-            # Persist the new MCP server set in the lockfile
-            MCPIntegrator.update_lockfile(new_mcp_servers)
+            # Persist the new MCP server set and configs in the lockfile
+            MCPIntegrator.update_lockfile(new_mcp_servers, mcp_configs=new_mcp_configs)
         elif should_install_mcp and not mcp_deps:
             # No MCP deps at all — remove any old APM-managed servers
             if old_mcp_servers:
                 MCPIntegrator.remove_stale(old_mcp_servers, runtime, exclude)
-                MCPIntegrator.update_lockfile(builtins.set())
+                MCPIntegrator.update_lockfile(builtins.set(), mcp_configs={})
             _rich_warning("No MCP dependencies found in apm.yml")
         elif not should_install_mcp and old_mcp_servers:
             # --only=apm: APM install regenerated the lockfile and dropped
             # mcp_servers.  Restore the previous set so it is not lost.
-            MCPIntegrator.update_lockfile(old_mcp_servers)
+            MCPIntegrator.update_lockfile(old_mcp_servers, mcp_configs=old_mcp_configs)
 
         # Show beautiful post-install summary
         _rich_blank_line()
@@ -645,6 +652,17 @@ def _collect_descendants(node, visited=None):
             _rich_info("No APM dependencies to install", symbol="check")
             return 0, 0, 0
 
+        # ------------------------------------------------------------------
+        # Orphan detection: packages in lockfile no longer in the manifest.
+        # Only relevant for a full install (not apm install <pkg>).
+        # We compute this NOW, before the download loop, so we know which old
+        # lockfile entries to remove from the merge and which deployed files
+        # to clean up after the loop.
+        # ------------------------------------------------------------------
+        intended_dep_keys: builtins.set = builtins.set(
+            d.get_unique_key() for d in deps_to_install
+        )
+
         # apm_modules directory already created above
 
         # Auto-detect target for integration (same logic as compile)
@@ -719,6 +737,14 @@ def _collect_descendants(node, visited=None):
         from apm_cli.integration.base_integrator import BaseIntegrator
         managed_files = BaseIntegrator.normalize_managed_files(managed_files)
 
+        # Collect deployed file paths for packages that are no longer in the manifest.
+        # detect_orphans() returns an empty set for partial installs automatically.
+        orphaned_deployed_files = detect_orphans(
+            existing_lockfile,
+            intended_dep_keys,
+            only_packages=only_packages,
+        )
+
         # Install each dependency with Rich progress display
         from rich.progress import (
             Progress,
@@ -744,8 +770,19 @@ def _collect_descendants(node, visited=None):
             # Skip if already downloaded during BFS resolution
             if _pd_key in callback_downloaded:
                 continue
+            # Detect if manifest ref changed from what's recorded in the lockfile.
+            # detect_ref_change() handles all transitions including None→ref.
+            _pd_locked_chk = (
+                existing_lockfile.get_dependency(_pd_key)
+                if existing_lockfile and not update_refs
+                else None
+            )
+            _pd_ref_changed = detect_ref_change(
+                _pd_ref, _pd_locked_chk, update_refs=update_refs
+            )
             # Skip if lockfile SHA matches local HEAD (Phase 5 check)
-            if _pd_path.exists() and existing_lockfile and not update_refs:
+            # — but only if the ref itself has not changed in the manifest.
+            if _pd_path.exists() and existing_lockfile and not update_refs and not _pd_ref_changed:
                 _pd_locked = existing_lockfile.get_dependency(_pd_key)
                 if _pd_locked and _pd_locked.resolved_commit and _pd_locked.resolved_commit != "cached":
                     try:
@@ -754,15 +791,11 @@ def _collect_descendants(node, visited=None):
                             continue
                     except Exception:
                         pass
-            # Build download ref (use locked commit for reproducibility)
-            _pd_dlref = str(_pd_ref)
-            if existing_lockfile and not update_refs:
-                _pd_locked = existing_lockfile.get_dependency(_pd_key)
-                if _pd_locked and _pd_locked.resolved_commit and _pd_locked.resolved_commit != "cached":
-                    _pd_base = _pd_ref.repo_url
-                    if _pd_ref.virtual_path:
-                        _pd_base = f"{_pd_base}/{_pd_ref.virtual_path}"
-                    _pd_dlref = f"{_pd_base}#{_pd_locked.resolved_commit}"
+            # Build download ref (use locked commit for reproducibility).
+            # build_download_ref() uses the manifest ref when ref_changed is True.
+            _pd_dlref = build_download_ref(
+                _pd_ref, existing_lockfile, update_refs=update_refs, ref_changed=_pd_ref_changed
+            )
             _need_download.append((_pd_ref, _pd_path, _pd_dlref))
 
         if _need_download and parallel_downloads > 0:
@@ -840,9 +873,20 @@ def _collect_descendants(node, visited=None):
                 ]
                 # Skip download if: already fetched by resolver callback, or cached tag/commit
                 already_resolved = dep_ref.get_unique_key() in callback_downloaded
+                # Detect if manifest ref changed vs what the lockfile recorded.
+                # detect_ref_change() handles all transitions including None→ref.
+                _dep_locked_chk = (
+                    existing_lockfile.get_dependency(dep_ref.get_unique_key())
+                    if existing_lockfile and not update_refs
+                    else None
+                )
+                ref_changed = detect_ref_change(
+                    dep_ref, _dep_locked_chk, update_refs=update_refs
+                )
                 # Phase 5 (#171): Also skip when lockfile SHA matches local HEAD
+                # — but not when the manifest ref has changed (user wants different version).
                 lockfile_match = False
-                if install_path.exists() and existing_lockfile and not update_refs:
+                if install_path.exists() and existing_lockfile and not update_refs and not ref_changed:
                     locked_dep = existing_lockfile.get_dependency(dep_ref.get_unique_key())
                     if locked_dep and locked_dep.resolved_commit and locked_dep.resolved_commit != "cached":
                         try:
@@ -1133,16 +1177,11 @@ def _collect_descendants(node, visited=None):
                         total=None,  # Indeterminate initially; git will update with actual counts
                     )
 
-                    # T5: Build download ref - use locked commit if available
-                    download_ref = str(dep_ref)
-                    if existing_lockfile and not update_refs:
-                        locked_dep = existing_lockfile.get_dependency(dep_ref.get_unique_key())
-                        if locked_dep and locked_dep.resolved_commit and locked_dep.resolved_commit != "cached":
-                            # Override with locked commit for reproducible install
-                            base_ref = dep_ref.repo_url
-                            if dep_ref.virtual_path:
-                                base_ref = f"{base_ref}/{dep_ref.virtual_path}"
-                            download_ref = f"{base_ref}#{locked_dep.resolved_commit}"
+                    # T5: Build download ref - use locked commit if available.
+                    # build_download_ref() uses manifest ref when ref_changed is True.
+                    download_ref = build_download_ref(
+                        dep_ref, existing_lockfile, update_refs=update_refs, ref_changed=ref_changed
+                    )
 
                     # Phase 4 (#171): Use pre-downloaded result if available
                     _dep_key = dep_ref.get_unique_key()
@@ -1388,6 +1427,45 @@ def _collect_descendants(node, visited=None):
         # Update .gitignore
         _update_gitignore_for_apm_modules()
 
+        # ------------------------------------------------------------------
+        # Orphan cleanup: remove deployed files for packages that were
+        # removed from the manifest.  This happens on every full install
+        # (no only_packages), making apm install idempotent with the manifest.
+        # Handles both regular files and directory entries (e.g., legacy skills).
+        # ------------------------------------------------------------------
+        if orphaned_deployed_files:
+            import shutil as _shutil
+            _removed_orphan_count = 0
+            _failed_orphan_count = 0
+            _deleted_orphan_paths: builtins.list = []
+            for _orphan_path in sorted(orphaned_deployed_files):
+                # validate_deploy_path() is the safety gate: it rejects path-traversal,
+                # requires .github/ or .claude/ prefix, and checks the resolved path
+                # stays within project_root — so rmtree is safe here.
+                if BaseIntegrator.validate_deploy_path(_orphan_path, project_root):
+                    _target = project_root / _orphan_path
+                    if _target.exists():
+                        try:
+                            if _target.is_dir():
+                                _shutil.rmtree(_target)
+                            else:
+                                _target.unlink()
+                            _deleted_orphan_paths.append(_target)
+                            _removed_orphan_count += 1
+                        except Exception as _orphan_err:
+                            _rich_warning(
+                                f"  └─ Could not remove orphaned path {_orphan_path}: {_orphan_err}"
+                            )
+                            _failed_orphan_count += 1
+            # Clean up empty parent directories left after file removal
+            if _deleted_orphan_paths:
+                BaseIntegrator.cleanup_empty_parents(_deleted_orphan_paths, project_root)
+            if _removed_orphan_count > 0:
+                _rich_info(
+                    f"Removed {_removed_orphan_count} file(s) from packages "
+                    "no longer in apm.yml"
+                )
+
         # Generate apm.lock for reproducible installs (T4: lockfile generation)
         if installed_packages:
             try:
@@ -1399,13 +1477,25 @@ def _collect_descendants(node, visited=None):
                 for dep_key, pkg_type in package_types.items():
                     if dep_key in lockfile.dependencies:
                         lockfile.dependencies[dep_key].package_type = pkg_type
-                # Merge with existing lockfile to preserve entries for packages
-                # not processed in this run (e.g. `apm install X` only installs X).
-                # Skip merge when update_refs is set — stale entries must not survive.
+                # Selectively merge entries from the existing lockfile:
+                #   - For partial installs (only_packages): preserve all old entries
+                #     (sequential install — only the specified package was processed).
+                #   - For full installs: only preserve entries for packages still in
+                #     the manifest that failed to download (in intended_dep_keys but
+                #     not in the new lockfile due to a download error).
+                #   - Orphaned entries (not in intended_dep_keys) are intentionally
+                #     dropped so the lockfile matches the manifest.
+                # Skip merge entirely when update_refs is set — stale entries must not survive.
                 if existing_lockfile and not update_refs:
                     for dep_key, dep in existing_lockfile.dependencies.items():
                         if dep_key not in lockfile.dependencies:
-                            lockfile.dependencies[dep_key] = dep
+                            if only_packages or dep_key in intended_dep_keys:
+                                # Preserve: partial install (sequential install support)
+                                # OR package still in manifest but failed to download.
+                                lockfile.dependencies[dep_key] = dep
+                            # else: orphan — package was in lockfile but is no longer in
+                            # the manifest (full install only). Don't preserve so the
+                            # lockfile stays in sync with what apm.yml declares.
                 lockfile_path = get_lockfile_path(project_root)
 
                 # When installing a subset of packages (apm install <pkg>),
diff --git a/src/apm_cli/deps/lockfile.py b/src/apm_cli/deps/lockfile.py
index daaba6ac..3fc2229e 100644
--- a/src/apm_cli/deps/lockfile.py
+++ b/src/apm_cli/deps/lockfile.py
@@ -123,6 +123,7 @@ class LockFile:
     apm_version: Optional[str] = None
     dependencies: Dict[str, LockedDependency] = field(default_factory=dict)
     mcp_servers: List[str] = field(default_factory=list)
+    mcp_configs: Dict[str, dict] = field(default_factory=dict)
 
     def add_dependency(self, dep: LockedDependency) -> None:
         """Add a dependency to the lock file."""
@@ -153,6 +154,8 @@ def to_yaml(self) -> str:
         data["dependencies"] = [dep.to_dict() for dep in self.get_all_dependencies()]
         if self.mcp_servers:
             data["mcp_servers"] = sorted(self.mcp_servers)
+        if self.mcp_configs:
+            data["mcp_configs"] = dict(sorted(self.mcp_configs.items()))
         return yaml.dump(
             data, default_flow_style=False, sort_keys=False, allow_unicode=True
         )
@@ -173,6 +176,7 @@ def from_yaml(cls, yaml_str: str) -> "LockFile":
         for dep_data in data.get("dependencies", []):
             lock.add_dependency(LockedDependency.from_dict(dep_data))
         lock.mcp_servers = list(data.get("mcp_servers", []))
+        lock.mcp_configs = dict(data.get("mcp_configs") or {})
         return lock
 
     def write(self, path: Path) -> None:
diff --git a/src/apm_cli/drift.py b/src/apm_cli/drift.py
new file mode 100644
index 00000000..555c977b
--- /dev/null
+++ b/src/apm_cli/drift.py
@@ -0,0 +1,199 @@
+"""Pure drift-detection helpers for diff-aware ``apm install``.
+
+These functions are stateless and side-effect-free, making them easy to test
+in isolation and to reuse from multiple call sites in ``install.py`` without
+duplicating logic.
+
+Three kinds of drift are detected:
+
+* **Ref drift** — the ``ref`` pinned in ``apm.yml`` differs from what the
+  lockfile recorded as ``resolved_ref``.  This includes transitions such as
+  ``None → "v1.0.0"`` (user adds a pin), ``"main" → None`` (user removes a
+  pin), ``"v1.0.0" → "v2.0.0"`` (user bumps the pin), and hash-based pins
+  (``None → "abc1234"`` or ``"abc1234" → "def5678"``).
+
+* **Orphan drift** — packages present in the lockfile but absent from the
+  current manifest.  Their deployed files should be removed.
+
+* **Config drift** — an already-installed dependency's serialised configuration
+  differs from the baseline stored in the lockfile.  (Currently only MCP
+  servers; extendable to other integrator types.)
+
+Scope / non-goals
+-----------------
+* **Hash-based refs** — handled identically to branch/tag refs: both
+  ``dep_ref.reference`` and ``locked_dep.resolved_ref`` store the raw ref
+  string from ``apm.yml``/lockfile respectively, so a change from
+  ``"abc1234"`` to ``"def5678"`` is detected just like ``"v1.0" → "v2.0"``.
+
+* **URL format changes** — transparent.  ``DependencyReference.parse()``
+  normalises all input formats (HTTPS, SSH, shorthand, FQDN) into the same
+  canonical ``repo_url`` before the lockfile stores them.  Changing
+  ``owner/repo`` to ``https://github.com/owner/repo.git`` in ``apm.yml`` is a
+  formatting-only change that produces the same unique key and is correctly
+  treated as no drift.
+
+* **Source/host changes** — *not* detected.  If a user changes the host of
+  an otherwise identical package (e.g. adding an enterprise FQDN prefix), the
+  unique key (``repo_url``, host-blind for the default host) may not change
+  and ``detect_ref_change()`` will not signal a re-download.  Host-level
+  changes require the user to ``apm remove`` + ``apm install`` the package, or
+  use ``--update``.
+"""
+
+from __future__ import annotations
+
+import builtins
+from pathlib import Path
+from typing import TYPE_CHECKING, Dict, List, Optional, Set
+
+if TYPE_CHECKING:
+    from apm_cli.deps.lockfile import LockFile, LockedDependency
+    from apm_cli.models.apm_package import DependencyReference
+
+
+# ---------------------------------------------------------------------------
+# Ref drift
+# ---------------------------------------------------------------------------
+
+def detect_ref_change(
+    dep_ref: "DependencyReference",
+    locked_dep: "Optional[LockedDependency]",
+    *,
+    update_refs: bool = False,
+) -> bool:
+    """Return ``True`` when the manifest ref differs from the locked resolved_ref.
+
+    Handles all transitions:
+
+    * ref *added*  (``None`` → ``"v1.0.0"``)
+    * ref *removed* (``"main"`` → ``None``)
+    * ref *changed* (``"v1.0.0"`` → ``"v2.0.0"``)
+
+    Args:
+        dep_ref: The dependency as declared in the current manifest.
+        locked_dep: The matching entry from the existing lockfile, or ``None``
+                    when the package is brand-new (not yet in the lockfile).
+        update_refs: Pass ``True`` when running in ``--update`` mode.  In that
+                     mode the lockfile is intentionally ignored, so this
+                     function always returns ``False`` to avoid double-action.
+
+    Returns:
+        ``True`` when a re-download is needed due to a ref change; ``False``
+        when the ref is unchanged, when the package is new, or when
+        ``update_refs=True``.
+    """
+    if update_refs:
+        return False
+    if locked_dep is None:
+        return False  # new package — not drift, just a first install
+    # Direct comparison: handles None→value, value→None, and value→value.
+    # No truthiness guard on locked_dep.resolved_ref — None != "v1.0.0" is True.
+    return dep_ref.reference != locked_dep.resolved_ref
+
+
+# ---------------------------------------------------------------------------
+# Orphan drift
+# ---------------------------------------------------------------------------
+
+def detect_orphans(
+    existing_lockfile: "Optional[LockFile]",
+    intended_dep_keys: builtins.set,
+    *,
+    only_packages: builtins.list,
+) -> builtins.set:
+    """Return the set of deployed file paths whose owning package left the manifest.
+
+    Only relevant for *full* installs (``only_packages`` is empty/None).
+    Partial installs (``apm install <pkg>``) preserve all existing lockfile
+    entries unchanged.
+
+    Args:
+        existing_lockfile: The lockfile from the previous install, or ``None``
+                           on first install.
+        intended_dep_keys: Set of unique dependency keys for packages declared
+                           in the updated manifest.
+        only_packages: When non-empty this is a partial install — return an
+                       empty set so no cleanup is performed.
+
+    Returns:
+        A set of workspace-relative path strings that belong to packages which
+        are no longer in the manifest.  The caller is responsible for actually
+        removing the files.
+    """
+    orphaned: builtins.set = builtins.set()
+    if only_packages or not existing_lockfile:
+        return orphaned
+    for dep_key, dep in existing_lockfile.dependencies.items():
+        if dep_key not in intended_dep_keys:
+            orphaned.update(dep.deployed_files)
+    return orphaned
+
+
+# ---------------------------------------------------------------------------
+# Config drift (integrator-agnostic)
+# ---------------------------------------------------------------------------
+
+def detect_config_drift(
+    current_configs: Dict[str, dict],
+    stored_configs: Dict[str, dict],
+) -> builtins.set:
+    """Return names of entries whose current config differs from the stored baseline.
+
+    Only entries that *have* a stored baseline and whose config has *changed*
+    are returned.  Brand-new entries (not in ``stored_configs``) are excluded
+    because they have never been installed — they are installs, not updates.
+
+    Args:
+        current_configs: Mapping of name → current serialised config (from the
+                         manifest / dependency objects).
+        stored_configs: Mapping of name → previously stored config (from the
+                        lockfile).
+
+    Returns:
+        A set of names (strings) whose configuration has drifted.
+    """
+    drifted: builtins.set = builtins.set()
+    for name, current in current_configs.items():
+        stored = stored_configs.get(name)
+        if stored is not None and stored != current:
+            drifted.add(name)
+    return drifted
+
+
+# ---------------------------------------------------------------------------
+# Download ref construction
+# ---------------------------------------------------------------------------
+
+def build_download_ref(
+    dep_ref: "DependencyReference",
+    existing_lockfile: "Optional[LockFile]",
+    *,
+    update_refs: bool,
+    ref_changed: bool,
+) -> str:
+    """Build the download-ref string passed to the package downloader.
+
+    Uses the locked commit SHA for reproducibility, unless:
+    * ``update_refs=True`` — intentional update run; use the manifest ref.
+    * ``ref_changed=True`` — the user changed the pin; use the manifest ref.
+
+    Args:
+        dep_ref: The dependency as declared in the current manifest.
+        existing_lockfile: Existing lockfile, or ``None`` on first install.
+        update_refs: Whether ``--update`` mode is active.
+        ref_changed: Whether :func:`detect_ref_change` returned ``True`` for
+                     this dependency.
+
+    Returns:
+        A ref string suitable for ``GitHubPackageDownloader.download_package``.
+    """
+    download_ref = str(dep_ref)
+    if existing_lockfile and not update_refs and not ref_changed:
+        locked_dep = existing_lockfile.get_dependency(dep_ref.get_unique_key())
+        if locked_dep and locked_dep.resolved_commit and locked_dep.resolved_commit != "cached":
+            base_ref = dep_ref.repo_url
+            if dep_ref.virtual_path:
+                base_ref = f"{base_ref}/{dep_ref.virtual_path}"
+            download_ref = f"{base_ref}#{locked_dep.resolved_commit}"
+    return download_ref
diff --git a/src/apm_cli/integration/mcp_integrator.py b/src/apm_cli/integration/mcp_integrator.py
index 7b9d8879..f30c4c0d 100644
--- a/src/apm_cli/integration/mcp_integrator.py
+++ b/src/apm_cli/integration/mcp_integrator.py
@@ -316,6 +316,53 @@ def get_server_names(mcp_deps: list) -> builtins.set:
                 names.add(dep)
         return names
 
+    @staticmethod
+    def get_server_configs(mcp_deps: list) -> builtins.dict:
+        """Extract server configs as {name: config_dict} from MCP dependencies."""
+        configs: builtins.dict = {}
+        for dep in mcp_deps:
+            if hasattr(dep, "to_dict") and hasattr(dep, "name"):
+                configs[dep.name] = dep.to_dict()
+            elif isinstance(dep, str):
+                configs[dep] = {"name": dep}
+        return configs
+
+    @staticmethod
+    def _append_drifted_to_install_list(
+        install_list: builtins.list,
+        drifted: builtins.set,
+    ) -> None:
+        """Append drifted server names to *install_list* without duplicates.
+
+        Appends in sorted order to guarantee deterministic CLI output.
+        Names already present in *install_list* are skipped.
+        """
+        existing = builtins.set(install_list)
+        for name in builtins.sorted(drifted):
+            if name not in existing:
+                install_list.append(name)
+
+    @staticmethod
+    def _detect_mcp_config_drift(
+        mcp_deps: list,
+        stored_configs: builtins.dict,
+    ) -> builtins.set:
+        """Return names of MCP deps whose manifest config differs from stored.
+
+        Compares each dependency's current serialized config against the
+        previously stored config in the lockfile.  Only dependencies that
+        have a stored baseline *and* whose config has changed are returned.
+        """
+        drifted: builtins.set = builtins.set()
+        for dep in mcp_deps:
+            if not hasattr(dep, "to_dict") or not hasattr(dep, "name"):
+                continue
+            current_config = dep.to_dict()
+            stored = stored_configs.get(dep.name)
+            if stored is not None and stored != current_config:
+                drifted.add(dep.name)
+        return drifted
+
     @staticmethod
     def _check_self_defined_servers_needing_installation(
         dep_names: list,
@@ -477,11 +524,19 @@ def remove_stale(
     def update_lockfile(
         mcp_server_names: builtins.set,
         lock_path: Optional[Path] = None,
+        *,
+        mcp_configs: Optional[builtins.dict] = None,
     ) -> None:
         """Update the lockfile with the current set of APM-managed MCP server names.
 
         Accepts the lock path directly to avoid a redundant disk read when the
         caller already has it.
+
+        Args:
+            mcp_server_names: Set of MCP server names to persist.
+            lock_path: Path to the lockfile.  Defaults to ``apm.lock`` in CWD.
+            mcp_configs: Keyword-only.  When provided, overwrites ``mcp_configs``
+                         in the lockfile (used for drift-detection baseline).
         """
         if lock_path is None:
             lock_path = Path.cwd() / "apm.lock"
@@ -492,6 +547,8 @@ def update_lockfile(
             if lockfile is None:
                 return
             lockfile.mcp_servers = sorted(mcp_server_names)
+            if mcp_configs is not None:
+                lockfile.mcp_configs = mcp_configs
             lockfile.save(lock_path)
         except Exception:
             logger.debug(
@@ -628,6 +685,7 @@ def install(
         exclude: str = None,
         verbose: bool = False,
         apm_config: dict = None,
+        stored_mcp_configs: dict = None,
     ) -> int:
         """Install MCP dependencies.
 
@@ -639,9 +697,12 @@ def install(
             verbose: Show detailed installation information.
             apm_config: The parsed apm.yml configuration dict (optional).
                 When not provided, the method loads it from disk.
+            stored_mcp_configs: Previously stored MCP configs from lockfile
+                for diff-aware installation.  When provided, servers whose
+                manifest config has changed are re-applied automatically.
 
         Returns:
-            Number of MCP servers newly configured.
+            Number of MCP servers newly configured or updated.
         """
         if not mcp_deps:
             _rich_warning("No MCP dependencies found in apm.yml")
@@ -668,6 +729,13 @@ def install(
         }
 
         console = _get_console()
+        # Track servers that were re-applied due to config drift
+        servers_to_update: builtins.set = builtins.set()
+        # Track successful updates separately so the summary counts are accurate
+        # even when some drift-detected servers fail to install.
+        successful_updates: builtins.set = builtins.set()
+        if stored_mcp_configs is None:
+            stored_mcp_configs = {}
 
         # Start MCP section with clean header
         if console:
@@ -839,12 +907,31 @@ def install(
                             target_runtimes, valid_servers
                         )
                     )
-                    already_configured_servers = [
+                    already_configured_candidates = [
                         dep
                         for dep in valid_servers
                         if dep not in servers_to_install
                     ]
 
+                    # Detect config drift for "already configured" servers
+                    if stored_mcp_configs and already_configured_candidates:
+                        drifted_reg_deps = [
+                            registry_dep_map[n]
+                            for n in already_configured_candidates
+                            if n in registry_dep_map
+                        ]
+                        drifted = MCPIntegrator._detect_mcp_config_drift(
+                            drifted_reg_deps, stored_mcp_configs,
+                        )
+                        if drifted:
+                            servers_to_update.update(drifted)
+                            MCPIntegrator._append_drifted_to_install_list(servers_to_install, drifted)
+                    already_configured_servers = [
+                        dep
+                        for dep in already_configured_candidates
+                        if dep not in servers_to_update
+                    ]
+
                     if not servers_to_install:
                         if console:
                             for dep in already_configured_servers:
@@ -905,10 +992,12 @@ def install(
 
                         # Install for each target runtime
                         for dep in servers_to_install:
+                            is_update = dep in servers_to_update
                             if console:
-                                console.print(f"│  [cyan]⬇️[/cyan]  {dep}")
+                                action = "↻" if is_update else "⬇️"
+                                console.print(f"│  [cyan]{action}[/cyan]  {dep}")
                                 console.print(
-                                    f"│     └─ Configuring for "
+                                    f"│     └─ {'Updating' if is_update else 'Configuring'} for "
                                     f"{', '.join([rt.title() for rt in target_runtimes])}..."
                                 )
 
@@ -927,11 +1016,15 @@ def install(
 
                             if any_ok:
                                 if console:
+                                    label = "updated" if is_update else "configured"
                                     console.print(
                                         f"│  [green]✓[/green]  {dep} → "
                                         f"{', '.join([rt.title() for rt in target_runtimes])}"
+                                        f" [dim]({label})[/dim]"
                                     )
                                 configured_count += 1
+                                if is_update:
+                                    successful_updates.add(dep)
                             elif console:
                                 console.print(
                                     f"│  [red]✗[/red]  {dep} — "
@@ -956,12 +1049,30 @@ def install(
                     target_runtimes,
                 )
             )
-            already_configured_self_defined = [
+            already_configured_candidates_sd = [
                 name
                 for name in self_defined_names
                 if name not in self_defined_to_install
             ]
 
+            # Detect config drift for "already configured" self-defined servers
+            if stored_mcp_configs and already_configured_candidates_sd:
+                drifted_sd_deps = [
+                    dep for dep in self_defined_deps
+                    if dep.name in already_configured_candidates_sd
+                ]
+                drifted_sd = MCPIntegrator._detect_mcp_config_drift(
+                    drifted_sd_deps, stored_mcp_configs,
+                )
+                if drifted_sd:
+                    servers_to_update.update(drifted_sd)
+                    MCPIntegrator._append_drifted_to_install_list(self_defined_to_install, drifted_sd)
+            already_configured_self_defined = [
+                name
+                for name in already_configured_candidates_sd
+                if name not in servers_to_update
+            ]
+
             if already_configured_self_defined:
                 if console:
                     for name in already_configured_self_defined:
@@ -983,18 +1094,20 @@ def install(
                 if dep.name not in self_defined_to_install:
                     continue
 
+                is_update = dep.name in servers_to_update
                 synthetic_info = MCPIntegrator._build_self_defined_info(dep)
                 self_defined_cache = {dep.name: synthetic_info}
                 self_defined_env = dep.env or {}
 
                 if console:
                     transport_label = dep.transport or "stdio"
+                    action = "↻" if is_update else "⬇️"
                     console.print(
-                        f"│  [cyan]⬇️[/cyan]  {dep.name} "
+                        f"│  [cyan]{action}[/cyan]  {dep.name} "
                         f"[dim](self-defined, {transport_label})[/dim]"
                     )
                     console.print(
-                        f"│     └─ Configuring for "
+                        f"│     └─ {'Updating' if is_update else 'Configuring'} for "
                         f"{', '.join([rt.title() for rt in target_runtimes])}..."
                     )
 
@@ -1012,11 +1125,15 @@ def install(
 
                 if any_ok:
                     if console:
+                        label = "updated" if is_update else "configured"
                         console.print(
                             f"│  [green]✓[/green]  {dep.name} → "
                             f"{', '.join([rt.title() for rt in target_runtimes])}"
+                            f" [dim]({label})[/dim]"
                         )
                     configured_count += 1
+                    if is_update:
+                        successful_updates.add(dep.name)
                 elif console:
                     console.print(
                         f"│  [red]✗[/red]  {dep.name} — "
@@ -1026,10 +1143,23 @@ def install(
         # Close the panel
         if console:
             if configured_count > 0:
-                console.print(
-                    f"└─ [green]Configured {configured_count} "
-                    f"server{'s' if configured_count != 1 else ''}[/green]"
-                )
+                # Use successful_updates (not servers_to_update) for accurate counts.
+                # servers_to_update = all drift-detected servers (some may have failed).
+                # successful_updates = servers that were re-applied AND succeeded.
+                update_count = builtins.len(successful_updates)
+                new_count = configured_count - update_count
+                parts = []
+                if new_count > 0:
+                    parts.append(
+                        f"configured {new_count} "
+                        f"server{'s' if new_count != 1 else ''}"
+                    )
+                if update_count > 0:
+                    parts.append(
+                        f"updated {update_count} "
+                        f"server{'s' if update_count != 1 else ''}"
+                    )
+                console.print(f"└─ [green]{', '.join(parts).capitalize()}[/green]")
             else:
                 console.print("└─ [green]All servers up to date[/green]")
 
diff --git a/tests/integration/test_diff_aware_install_e2e.py b/tests/integration/test_diff_aware_install_e2e.py
new file mode 100644
index 00000000..ec176379
--- /dev/null
+++ b/tests/integration/test_diff_aware_install_e2e.py
@@ -0,0 +1,325 @@
+"""End-to-end integration tests for diff-aware apm install.
+
+Tests the complete manifest-as-source-of-truth lifecycle with real packages:
+- Package removed from apm.yml: apm install cleans up deployed files and lockfile
+- Package ref/version changed in apm.yml: apm install re-downloads without --update
+- MCP config drift: apm install re-applies changed MCP server config (unit-tested;
+  omitted from e2e since it requires a real runtime to be configured)
+
+Requires network access and GITHUB_TOKEN/GITHUB_APM_PAT for GitHub API.
+Uses real packages from GitHub:
+  - microsoft/apm-sample-package (deployed prompts, agents, etc.)
+"""
+
+import os
+import shutil
+import subprocess
+
+import pytest
+import yaml
+from pathlib import Path
+
+
+# Skip all tests if no GitHub token is available
+pytestmark = pytest.mark.skipif(
+    not os.environ.get("GITHUB_APM_PAT") and not os.environ.get("GITHUB_TOKEN"),
+    reason="GITHUB_APM_PAT or GITHUB_TOKEN required for GitHub API access",
+)
+
+
+@pytest.fixture
+def apm_command():
+    """Get the path to the APM CLI executable."""
+    apm_on_path = shutil.which("apm")
+    if apm_on_path:
+        return apm_on_path
+    venv_apm = Path(__file__).parent.parent.parent / ".venv" / "bin" / "apm"
+    if venv_apm.exists():
+        return str(venv_apm)
+    return "apm"
+
+
+@pytest.fixture
+def temp_project(tmp_path):
+    """Create a temporary APM project with .github/ for VSCode target detection."""
+    project_dir = tmp_path / "diff-aware-install-test"
+    project_dir.mkdir()
+    (project_dir / ".github").mkdir()
+    return project_dir
+
+
+def _run_apm(apm_command, args, cwd, timeout=180):
+    """Run an apm CLI command and return the result."""
+    return subprocess.run(
+        [apm_command] + args,
+        cwd=cwd,
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+    )
+
+
+def _write_apm_yml(project_dir, packages):
+    """Write apm.yml with the given list of APM package specs."""
+    config = {
+        "name": "diff-aware-test",
+        "version": "1.0.0",
+        "dependencies": {
+            "apm": packages,
+            "mcp": [],
+        },
+    }
+    (project_dir / "apm.yml").write_text(
+        yaml.dump(config, default_flow_style=False), encoding="utf-8"
+    )
+
+
+def _read_lockfile(project_dir):
+    """Read and parse apm.lock from the project directory."""
+    lock_path = project_dir / "apm.lock"
+    if not lock_path.exists():
+        return None
+    with open(lock_path, encoding="utf-8") as f:
+        return yaml.safe_load(f)
+
+
+def _get_locked_dep(lockfile, repo_url):
+    """Get a dependency entry from lockfile by repo_url."""
+    if not lockfile or "dependencies" not in lockfile:
+        return None
+    deps = lockfile["dependencies"]
+    if isinstance(deps, list):
+        for entry in deps:
+            if entry.get("repo_url") == repo_url:
+                return entry
+    return None
+
+
+def _collect_deployed_files(project_dir, dep_entry):
+    """Return existing deployed files from a lockfile dep entry."""
+    if not dep_entry or not dep_entry.get("deployed_files"):
+        return []
+    return [f for f in dep_entry["deployed_files"] if (project_dir / f).exists()]
+
+
+# ---------------------------------------------------------------------------
+# Scenario 1: Package removed from manifest — apm install cleans up
+# ---------------------------------------------------------------------------
+
+class TestPackageRemovedFromManifest:
+    """When a package is removed from apm.yml, apm install should clean up
+    its deployed files and remove it from the lockfile."""
+
+    def test_removed_package_files_cleaned_on_install(self, temp_project, apm_command):
+        """Files deployed by a removed package disappear on the next apm install."""
+        # ── Step 1: install the package ──
+        _write_apm_yml(temp_project, ["microsoft/apm-sample-package"])
+        result1 = _run_apm(apm_command, ["install"], temp_project)
+        assert result1.returncode == 0, (
+            f"Initial install failed:\nSTDOUT: {result1.stdout}\nSTDERR: {result1.stderr}"
+        )
+
+        # ── Step 2: verify deployed files exist and are tracked ──
+        lockfile_before = _read_lockfile(temp_project)
+        assert lockfile_before is not None, "apm.lock was not created"
+        dep_before = _get_locked_dep(lockfile_before, "microsoft/apm-sample-package")
+        assert dep_before is not None, "Package not in lockfile after install"
+        deployed_before = _collect_deployed_files(temp_project, dep_before)
+        assert len(deployed_before) > 0, (
+            "No deployed files found on disk — cannot verify cleanup"
+        )
+
+        # ── Step 3: remove the package from manifest ──
+        _write_apm_yml(temp_project, [])
+
+        # ── Step 4: run apm install (no packages) — should detect orphan ──
+        result2 = _run_apm(apm_command, ["install", "--only=apm"], temp_project)
+        assert result2.returncode == 0, (
+            f"Install after removal failed:\nSTDOUT: {result2.stdout}\nSTDERR: {result2.stderr}"
+        )
+
+        # ── Step 5: verify deployed files are gone ──
+        for rel_path in deployed_before:
+            full_path = temp_project / rel_path
+            assert not full_path.exists(), (
+                f"Orphaned file {rel_path} was NOT cleaned up by apm install"
+            )
+
+    def test_removed_package_absent_from_lockfile_after_install(self, temp_project, apm_command):
+        """After removing a package from apm.yml, apm install removes it from lockfile."""
+        # ── Install ──
+        _write_apm_yml(temp_project, ["microsoft/apm-sample-package"])
+        result1 = _run_apm(apm_command, ["install"], temp_project)
+        assert result1.returncode == 0, (
+            f"Initial install failed:\nSTDOUT: {result1.stdout}\nSTDERR: {result1.stderr}"
+        )
+
+        # ── Remove from manifest ──
+        _write_apm_yml(temp_project, [])
+
+        # ── Re-install ──
+        result2 = _run_apm(apm_command, ["install", "--only=apm"], temp_project)
+        assert result2.returncode == 0, (
+            f"Install after removal failed:\nSTDOUT: {result2.stdout}\nSTDERR: {result2.stderr}"
+        )
+
+        # ── Verify lockfile no longer has the removed package ──
+        lockfile_after = _read_lockfile(temp_project)
+        if lockfile_after and lockfile_after.get("dependencies"):
+            dep_after = _get_locked_dep(lockfile_after, "microsoft/apm-sample-package")
+            assert dep_after is None, (
+                "Removed package still present in apm.lock after apm install"
+            )
+
+    def test_remaining_package_unaffected_by_removal(self, temp_project, apm_command):
+        """Files from packages still in the manifest are untouched."""
+        # ── Install two packages ──
+        _write_apm_yml(temp_project, [
+            "microsoft/apm-sample-package",
+            "github/awesome-copilot/skills/aspire",
+        ])
+        result1 = _run_apm(apm_command, ["install"], temp_project)
+        assert result1.returncode == 0, (
+            f"Initial install failed:\nSTDOUT: {result1.stdout}\nSTDERR: {result1.stderr}"
+        )
+
+        lockfile_before = _read_lockfile(temp_project)
+        sample_dep = _get_locked_dep(lockfile_before, "microsoft/apm-sample-package")
+        if not sample_dep or not _collect_deployed_files(temp_project, sample_dep):
+            pytest.skip("apm-sample-package deployed no files, cannot verify")
+
+        # ── Remove only apm-sample-package ──
+        _write_apm_yml(temp_project, ["github/awesome-copilot/skills/aspire"])
+        result2 = _run_apm(apm_command, ["install", "--only=apm"], temp_project)
+        assert result2.returncode == 0, (
+            f"Second install failed:\nSTDOUT: {result2.stdout}\nSTDERR: {result2.stderr}"
+        )
+
+        # ── apm-sample-package files should be gone ──
+        for rel_path in (sample_dep.get("deployed_files") or []):
+            # The files that were deployed should no longer exist
+            assert not (temp_project / rel_path).exists(), (
+                f"Removed package file {rel_path} still on disk"
+            )
+
+
+# ---------------------------------------------------------------------------
+# Scenario 2: Package ref changed — apm install re-downloads
+# ---------------------------------------------------------------------------
+
+class TestPackageRefChangedInManifest:
+    """When the ref in apm.yml changes, apm install re-downloads without --update."""
+
+    def test_ref_change_triggers_re_download(self, temp_project, apm_command):
+        """Changing the ref in apm.yml from one value to another causes re-download."""
+        # ── Step 1: install with an explicit commit-pinned ref ──
+        # We install first without a ref (using default branch), so the lockfile
+        # records the resolved_ref as the default branch or latest commit.
+        _write_apm_yml(temp_project, ["microsoft/apm-sample-package"])
+        result1 = _run_apm(apm_command, ["install"], temp_project)
+        assert result1.returncode == 0, (
+            f"Initial install failed:\nSTDOUT: {result1.stdout}\nSTDERR: {result1.stderr}"
+        )
+
+        lockfile1 = _read_lockfile(temp_project)
+        assert lockfile1 is not None, "apm.lock was not created"
+        dep1 = _get_locked_dep(lockfile1, "microsoft/apm-sample-package")
+        assert dep1 is not None, "Package not in lockfile"
+        original_commit = dep1.get("resolved_commit")
+        assert original_commit, "No resolved_commit in lockfile after install"
+
+        # ── Step 2: change ref to "main" explicitly (from unset → explicit branch) ──
+        # This differs from the lockfile's resolved_ref (which may be None/default).
+        # For the test to be meaningful we pick a known ref that EXISTS in the repo.
+        # We use "main" — the primary branch — which definitely exists.
+        _write_apm_yml(temp_project, [
+            {"git": "https://github.com/microsoft/apm-sample-package.git", "ref": "main"}
+        ])
+
+        # ── Step 3: run install WITHOUT --update ──
+        result2 = _run_apm(apm_command, ["install", "--only=apm"], temp_project)
+        assert result2.returncode == 0, (
+            f"Install with changed ref failed:\nSTDOUT: {result2.stdout}\nSTDERR: {result2.stderr}"
+        )
+
+        # ── Step 4: verify the package was re-processed ──
+        # Even if the commit hash is the same (main hasn't changed), the install
+        # must not silently skip the package — it must re-evaluate the ref.
+        # We verify the lockfile was updated and the package directory still exists.
+        lockfile2 = _read_lockfile(temp_project)
+        assert lockfile2 is not None, "apm.lock missing after second install"
+        dep2 = _get_locked_dep(lockfile2, "microsoft/apm-sample-package")
+        assert dep2 is not None, "Package disappeared from lockfile after ref change"
+
+        # The re-download should write back to lockfile; package dir must exist
+        package_dir = temp_project / "apm_modules" / "microsoft" / "apm-sample-package"
+        assert package_dir.exists(), (
+            "Package directory disappeared after re-download for ref change"
+        )
+
+    def test_no_ref_change_does_not_re_download(self, temp_project, apm_command):
+        """Without a ref change, apm install uses the lockfile SHA (idempotent)."""
+        # ── Install ──
+        _write_apm_yml(temp_project, ["microsoft/apm-sample-package"])
+        result1 = _run_apm(apm_command, ["install"], temp_project)
+        assert result1.returncode == 0, (
+            f"Initial install failed:\nSTDOUT: {result1.stdout}\nSTDERR: {result1.stderr}"
+        )
+
+        lockfile1 = _read_lockfile(temp_project)
+        dep1 = _get_locked_dep(lockfile1, "microsoft/apm-sample-package")
+        commit_before = dep1.get("resolved_commit") if dep1 else None
+
+        # ── Re-install without changing the ref ──
+        result2 = _run_apm(apm_command, ["install", "--only=apm"], temp_project)
+        assert result2.returncode == 0, (
+            f"Re-install failed:\nSTDOUT: {result2.stdout}\nSTDERR: {result2.stderr}"
+        )
+
+        # ── Commit should remain the same (lockfile pinned) ──
+        lockfile2 = _read_lockfile(temp_project)
+        dep2 = _get_locked_dep(lockfile2, "microsoft/apm-sample-package")
+        commit_after = dep2.get("resolved_commit") if dep2 else None
+
+        if commit_before and commit_after:
+            assert commit_before == commit_after, (
+                f"Lockfile SHA changed without a ref change: "
+                f"{commit_before} → {commit_after}"
+            )
+
+
+# ---------------------------------------------------------------------------
+# Scenario 3: Full install is idempotent when manifest unchanged
+# ---------------------------------------------------------------------------
+
+class TestFullInstallIdempotent:
+    """Running apm install multiple times without manifest changes is safe."""
+
+    def test_repeated_install_does_not_remove_files(self, temp_project, apm_command):
+        """Repeated apm install with same manifest preserves deployed files."""
+        _write_apm_yml(temp_project, ["microsoft/apm-sample-package"])
+
+        result1 = _run_apm(apm_command, ["install"], temp_project)
+        assert result1.returncode == 0, (
+            f"First install failed:\nSTDOUT: {result1.stdout}\nSTDERR: {result1.stderr}"
+        )
+
+        lockfile1 = _read_lockfile(temp_project)
+        dep1 = _get_locked_dep(lockfile1, "microsoft/apm-sample-package")
+        files_before = dep1.get("deployed_files", []) if dep1 else []
+
+        result2 = _run_apm(apm_command, ["install"], temp_project)
+        assert result2.returncode == 0, (
+            f"Second install failed:\nSTDOUT: {result2.stdout}\nSTDERR: {result2.stderr}"
+        )
+
+        # All files from the first install must still exist
+        for rel_path in files_before:
+            assert (temp_project / rel_path).exists(), (
+                f"File {rel_path} disappeared after idempotent re-install"
+            )
+
+        # Package must still be in lockfile
+        lockfile2 = _read_lockfile(temp_project)
+        dep2 = _get_locked_dep(lockfile2, "microsoft/apm-sample-package")
+        assert dep2 is not None, "Package missing from lockfile after idempotent re-install"
diff --git a/tests/test_lockfile.py b/tests/test_lockfile.py
index 9f77cf28..c4106537 100644
--- a/tests/test_lockfile.py
+++ b/tests/test_lockfile.py
@@ -99,6 +99,65 @@ def test_mcp_servers_from_yaml(self):
         lock = LockFile.from_yaml(yaml_str)
         assert lock.mcp_servers == ["github", "acme-kb"]
 
+    def test_mcp_configs_round_trip(self, tmp_path):
+        """mcp_configs survive a write/read cycle."""
+        lock = LockFile()
+        lock.mcp_configs = {
+            "github": {"name": "github", "transport": "stdio"},
+            "internal-kb": {
+                "name": "internal-kb",
+                "registry": False,
+                "transport": "http",
+                "url": "https://kb.example.com",
+            },
+        }
+        lock_path = tmp_path / "apm.lock"
+        lock.write(lock_path)
+
+        loaded = LockFile.read(lock_path)
+        assert loaded is not None
+        assert loaded.mcp_configs == lock.mcp_configs
+
+    def test_mcp_configs_empty_by_default(self):
+        lock = LockFile()
+        assert lock.mcp_configs == {}
+        yaml_str = lock.to_yaml()
+        assert "mcp_configs" not in yaml_str  # omitted when empty
+
+    def test_mcp_configs_from_yaml(self):
+        yaml_str = (
+            'lockfile_version: "1"\n'
+            'dependencies: []\n'
+            'mcp_configs:\n'
+            '  github:\n'
+            '    name: github\n'
+            '    transport: stdio\n'
+        )
+        lock = LockFile.from_yaml(yaml_str)
+        assert lock.mcp_configs == {"github": {"name": "github", "transport": "stdio"}}
+
+    def test_mcp_configs_backward_compat_missing(self):
+        """Old lockfiles without mcp_configs should get an empty dict."""
+        yaml_str = (
+            'lockfile_version: "1"\n'
+            'dependencies: []\n'
+            'mcp_servers:\n'
+            '  - github\n'
+        )
+        lock = LockFile.from_yaml(yaml_str)
+        assert lock.mcp_servers == ["github"]
+        assert lock.mcp_configs == {}
+
+    def test_mcp_configs_backward_compat_null(self):
+        """Lockfiles with mcp_configs: (null) should get an empty dict, not raise TypeError."""
+        yaml_str = (
+            'lockfile_version: "1"\n'
+            'dependencies: []\n'
+            'mcp_configs:\n'  # YAML null value
+        )
+        lock = LockFile.from_yaml(yaml_str)
+        assert lock.mcp_configs == {}
+
     def test_read_nonexistent(self, tmp_path):
         loaded = LockFile.read(tmp_path / "apm.lock")
         assert loaded is None
diff --git a/tests/unit/test_install_update.py b/tests/unit/test_install_update.py
index f7934a72..6a92cab9 100644
--- a/tests/unit/test_install_update.py
+++ b/tests/unit/test_install_update.py
@@ -2,10 +2,16 @@
 
 Verifies that `apm install --update` bypasses lockfile-pinned SHAs
 and re-fetches the latest content, especially for subdirectory packages.
+
+Also tests the drift detection helpers in ``apm_cli/drift.py``:
+- ``detect_ref_change()`` covers all ref transition cases (None→value, etc.)
+- ``detect_orphans()`` covers full vs partial install
+- ``build_download_ref()`` validates locked-SHA vs manifest-ref selection
 """
 
 from unittest.mock import Mock
 
+from apm_cli.drift import build_download_ref, detect_config_drift, detect_orphans, detect_ref_change
 from apm_cli.models.apm_package import DependencyReference
 
 
@@ -236,3 +242,248 @@ def test_pre_download_lockfile_override_without_update(self):
 
         ref = self._build_pre_download_ref(dep, lockfile, update_refs=False)
         assert "#abc123def456" in ref
+
+
+class TestRefChangedDetection:
+    """Tests for detect_ref_change() in drift.py.
+
+    When the user changes the ref pin in apm.yml (e.g., from v1.0.0 to v2.0.0),
+    apm install should detect the drift and re-download without --update.
+
+    Key improvement over the old inline logic: handles all None transitions.
+    """
+
+    def _make_dep(self, reference):
+        return DependencyReference(
+            repo_url="owner/repo",
+            host="github.com",
+            reference=reference,
+        )
+
+    def _mock_locked_dep(self, resolved_ref, resolved_commit="abc123"):
+        locked_dep = Mock()
+        locked_dep.resolved_ref = resolved_ref
+        locked_dep.resolved_commit = resolved_commit
+        return locked_dep
+
+    def test_no_drift_when_refs_match(self):
+        """No drift when manifest ref matches lockfile resolved_ref."""
+        dep = self._make_dep("v1.0.0")
+        locked = self._mock_locked_dep("v1.0.0")
+        assert detect_ref_change(dep, locked) is False
+
+    def test_drift_when_ref_changed(self):
+        """Drift detected when manifest ref changed from v1.0.0 to v2.0.0."""
+        dep = self._make_dep("v2.0.0")
+        locked = self._mock_locked_dep("v1.0.0")
+        assert detect_ref_change(dep, locked) is True
+
+    def test_drift_when_ref_added(self):
+        """Drift detected when ref added (None → 'v1.0.0').
+
+        This was a false-negative in the old inline logic because of the
+        ``and locked_dep.resolved_ref`` guard. drift.py removes that guard.
+        """
+        dep = self._make_dep("v1.0.0")
+        locked = self._mock_locked_dep(None)  # package was installed without a ref
+        assert detect_ref_change(dep, locked) is True
+
+    def test_drift_when_ref_removed(self):
+        """Drift detected when ref removed ('main' → None)."""
+        dep = self._make_dep(None)
+        locked = self._mock_locked_dep("main")
+        assert detect_ref_change(dep, locked) is True
+
+    def test_no_drift_when_both_refs_none(self):
+        """No drift when both manifest and lockfile have no ref."""
+        dep = self._make_dep(None)
+        locked = self._mock_locked_dep(None)
+        assert detect_ref_change(dep, locked) is False
+
+    def test_no_drift_when_no_locked_dep(self):
+        """No drift when locked_dep is None (new package, first install)."""
+        dep = self._make_dep("v1.0.0")
+        assert detect_ref_change(dep, None) is False
+
+    def test_no_drift_when_update_refs(self):
+        """With update_refs=True, always returns False (--update mode)."""
+        dep = self._make_dep("v2.0.0")
+        locked = self._mock_locked_dep("v1.0.0")
+        assert detect_ref_change(dep, locked, update_refs=True) is False
+
+    def test_build_download_ref_uses_new_ref_when_changed(self):
+        """When ref changed, build_download_ref does NOT use locked commit SHA."""
+        dep = self._make_dep("v2.0.0")
+        lockfile = Mock()
+        locked_dep = self._mock_locked_dep("v1.0.0", "abc123")
+        lockfile.get_dependency = Mock(return_value=locked_dep)
+        ref_changed = detect_ref_change(dep, locked_dep)
+        assert ref_changed is True
+        download_ref = build_download_ref(
+            dep, lockfile, update_refs=False, ref_changed=ref_changed
+        )
+        assert "#abc123" not in download_ref
+
+    def test_build_download_ref_uses_locked_sha_when_no_change(self):
+        """When ref unchanged, build_download_ref uses the locked commit SHA."""
+        dep = self._make_dep("v1.0.0")
+        lockfile = Mock()
+        locked_dep = self._mock_locked_dep("v1.0.0", "abc123")
+        lockfile.get_dependency = Mock(return_value=locked_dep)
+        ref_changed = detect_ref_change(dep, locked_dep)
+        assert ref_changed is False
+        download_ref = build_download_ref(
+            dep, lockfile, update_refs=False, ref_changed=ref_changed
+        )
+        assert "#abc123" in download_ref
+
+
+class TestOrphanDeployedFilesDetection:
+    """Tests for detect_orphans() in drift.py.
+
+    When packages are removed from apm.yml and apm install is run (full install),
+    the deployed files should be identified for cleanup.
+    """
+
+    @staticmethod
+    def _should_merge_lockfile_entry(dep_key, lockfile_dependencies, only_packages, intended_dep_keys):
+        """Reproduce the lockfile merge condition from install.py.
+
+        Returns True if the dep_key should be merged into the new lockfile.
+        Logic: only merge if (a) not already in new lockfile AND
+               (b) either partial install OR package still in intended set.
+        """
+        if dep_key in lockfile_dependencies:
+            return False  # already in new lockfile
+        return bool(only_packages or dep_key in intended_dep_keys)
+
+    def _mock_lockfile_with_deps(self, deps):
+        """Build a mock lockfile with {dep_key: [deployed_files]} entries."""
+        lockfile = Mock()
+        dep_mocks = {}
+        for dep_key, deployed_files in deps.items():
+            dep = Mock()
+            dep.deployed_files = deployed_files
+            dep_mocks[dep_key] = dep
+        lockfile.dependencies = dep_mocks
+        return lockfile
+
+    def test_no_orphans_when_all_packages_still_in_manifest(self):
+        """No orphaned files when all lockfile packages are still in manifest."""
+        lockfile = self._mock_lockfile_with_deps({
+            "owner/pkg-a": [".github/prompts/a.prompt.md"],
+            "owner/pkg-b": [".github/prompts/b.prompt.md"],
+        })
+        intended = {"owner/pkg-a", "owner/pkg-b"}
+        orphans = detect_orphans(lockfile, intended, only_packages=None)
+        assert orphans == set()
+
+    def test_orphaned_files_when_package_removed(self):
+        """Deployed files for removed package should be detected as orphans."""
+        lockfile = self._mock_lockfile_with_deps({
+            "owner/pkg-a": [".github/prompts/a.prompt.md"],
+            "owner/pkg-removed": [
+                ".github/prompts/removed.prompt.md",
+                ".github/instructions/removed.instructions.md",
+            ],
+        })
+        intended = {"owner/pkg-a"}  # pkg-removed not in new manifest
+        orphans = detect_orphans(lockfile, intended, only_packages=None)
+        assert orphans == {
+            ".github/prompts/removed.prompt.md",
+            ".github/instructions/removed.instructions.md",
+        }
+
+    def test_no_orphans_for_partial_install(self):
+        """Orphan detection is skipped for partial installs (only_packages)."""
+        lockfile = self._mock_lockfile_with_deps({
+            "owner/pkg-a": [".github/prompts/a.prompt.md"],
+            "owner/pkg-removed": [".github/prompts/removed.prompt.md"],
+        })
+        intended = {"owner/pkg-a"}
+        orphans = detect_orphans(lockfile, intended, only_packages=["owner/pkg-a"])
+        assert orphans == set()
+
+    def test_no_orphans_when_no_lockfile(self):
+        """No orphaned files when there is no existing lockfile."""
+        orphans = detect_orphans(None, {"owner/pkg-a"}, only_packages=None)
+        assert orphans == set()
+
+    def test_lockfile_merge_drops_orphan_in_full_install(self):
+        """In a full install, orphaned lockfile entries should NOT be merged."""
+        # Simulate: new lockfile has pkg-a, old lockfile has pkg-a + pkg-removed
+        new_lockfile_deps = {"owner/pkg-a"}
+        intended = {"owner/pkg-a"}  # pkg-removed no longer in manifest
+
+        # pkg-removed should NOT be merged (orphan)
+        assert not self._should_merge_lockfile_entry(
+            "owner/pkg-removed", new_lockfile_deps, only_packages=None, intended_dep_keys=intended
+        )
+
+    def test_lockfile_merge_preserves_failed_download_in_full_install(self):
+        """In a full install, failed downloads (still in manifest) should be preserved."""
+        new_lockfile_deps = {"owner/pkg-a"}  # pkg-b failed to download
+        intended = {"owner/pkg-a", "owner/pkg-b"}  # both in manifest
+
+        # pkg-b should be preserved (still in manifest, just failed)
+        assert self._should_merge_lockfile_entry(
+            "owner/pkg-b", new_lockfile_deps, only_packages=None, intended_dep_keys=intended
+        )
+
+    def test_lockfile_merge_preserves_all_for_partial_install(self):
+        """For partial installs, ALL old lockfile entries should be preserved."""
+        new_lockfile_deps = {"owner/pkg-a"}
+        intended = {"owner/pkg-a"}  # pkg-removed not in new manifest
+
+        # pkg-removed should STILL be preserved in a partial install
+        assert self._should_merge_lockfile_entry(
+            "owner/pkg-removed", new_lockfile_deps, only_packages=["owner/pkg-a"], intended_dep_keys=intended
+        )
+
+
+class TestDetectConfigDrift:
+    """Tests for detect_config_drift() in drift.py.
+
+    Config drift means an already-installed item's serialized config
+    in apm.yml differs from the stored baseline in the lockfile.
+    """
+
+    def test_no_drift_when_configs_match(self):
+        """No drift when current config is identical to stored config."""
+        current = {"name": "my-server", "url": "http://example.com"}
+        stored = {"my-server": {"name": "my-server", "url": "http://example.com"}}
+        assert detect_config_drift({"my-server": current}, stored) == set()
+
+    def test_drift_when_config_changed(self):
+        """Drift detected when config value changed."""
+        current = {"name": "my-server", "url": "http://new.example.com"}
+        stored = {"my-server": {"name": "my-server", "url": "http://old.example.com"}}
+        assert detect_config_drift({"my-server": current}, stored) == {"my-server"}
+
+    def test_no_drift_for_new_entry_without_baseline(self):
+        """Brand-new entry without a stored baseline is NOT drift — it's a first install."""
+        current = {"name": "brand-new", "url": "http://example.com"}
+        assert detect_config_drift({"brand-new": current}, {}) == set()
+
+    def test_drift_when_env_changed(self):
+        """Drift detected when env variables change."""
+        current = {"name": "s", "env": {"TOKEN": "new"}}
+        stored = {"s": {"name": "s", "env": {"TOKEN": "old"}}}
+        assert detect_config_drift({"s": current}, stored) == {"s"}
+
+    def test_no_drift_when_stored_configs_empty(self):
+        """No drift when no stored baseline exists (backward compat)."""
+        current = {"name": "s", "url": "http://x.com"}
+        assert detect_config_drift({"s": current}, {}) == set()
+
+    def test_multiple_entries_partial_drift(self):
+        """Only changed entries are reported."""
+        current_configs = {
+            "unchanged": {"url": "http://a.com"},
+            "changed": {"url": "http://new.com"},
+        }
+        stored = {
+            "unchanged": {"url": "http://a.com"},
+            "changed": {"url": "http://old.com"},
+        }
+        assert detect_config_drift(current_configs, stored) == {"changed"}
diff --git a/tests/unit/test_transitive_mcp.py b/tests/unit/test_transitive_mcp.py
index dea4bb9f..c51b8e70 100644
--- a/tests/unit/test_transitive_mcp.py
+++ b/tests/unit/test_transitive_mcp.py
@@ -708,3 +708,271 @@ def test_mixed_self_defined_shows_already_configured(
         )
         assert "existing-srv" in printed_lines
         assert "already configured" in printed_lines
+
+
+# ---------------------------------------------------------------------------
+# _detect_mcp_config_drift
+# ---------------------------------------------------------------------------
+class TestDetectMCPConfigDrift:
+    """Tests for config drift detection between manifest and lockfile."""
+
+    def test_no_drift_when_configs_match(self):
+        """No drift when manifest config matches stored config."""
+        dep = MCPDependency(name="github", transport="stdio")
+        stored = {"github": {"name": "github", "transport": "stdio"}}
+        result = MCPIntegrator._detect_mcp_config_drift([dep], stored)
+        assert result == set()
+
+    def test_drift_detected_when_env_changes(self):
+        """Drift detected when env vars change."""
+        dep = MCPDependency(
+            name="github", transport="stdio", env={"TOKEN": "new-value"}
+        )
+        stored = {
+            "github": {"name": "github", "transport": "stdio", "env": {"TOKEN": "old-value"}}
+        }
+        result = MCPIntegrator._detect_mcp_config_drift([dep], stored)
+        assert result == {"github"}
+
+    def test_drift_detected_when_url_changes(self):
+        """Drift detected when URL changes for self-defined server."""
+        dep = MCPDependency(
+            name="internal-kb", registry=False, transport="http",
+            url="https://new-kb.example.com",
+        )
+        stored = {
+            "internal-kb": {
+                "name": "internal-kb", "registry": False,
+                "transport": "http", "url": "https://old-kb.example.com",
+            }
+        }
+        result = MCPIntegrator._detect_mcp_config_drift([dep], stored)
+        assert result == {"internal-kb"}
+
+    def test_drift_detected_when_transport_changes(self):
+        """Drift detected when transport type changes."""
+        dep = MCPDependency(name="github", transport="stdio")
+        stored = {"github": {"name": "github", "transport": "sse"}}
+        result = MCPIntegrator._detect_mcp_config_drift([dep], stored)
+        assert result == {"github"}
+
+    def test_drift_detected_when_args_change(self):
+        """Drift detected when args change."""
+        dep = MCPDependency(
+            name="github", transport="stdio", args=["--new-flag"]
+        )
+        stored = {"github": {"name": "github", "transport": "stdio", "args": ["--old-flag"]}}
+        result = MCPIntegrator._detect_mcp_config_drift([dep], stored)
+        assert result == {"github"}
+
+    def test_drift_detected_when_tools_change(self):
+        """Drift detected when tools list changes."""
+        dep = MCPDependency(name="github", tools=["repos", "issues"])
+        stored = {"github": {"name": "github", "tools": ["repos"]}}
+        result = MCPIntegrator._detect_mcp_config_drift([dep], stored)
+        assert result == {"github"}
+
+    def test_no_drift_when_server_not_in_stored(self):
+        """No drift when server has no stored baseline."""
+        dep = MCPDependency(name="new-server", transport="stdio")
+        stored = {"other-server": {"name": "other-server"}}
+        result = MCPIntegrator._detect_mcp_config_drift([dep], stored)
+        assert result == set()
+
+    def test_no_drift_with_empty_stored_configs(self):
+        """No drift when stored configs are empty (first install)."""
+        dep = MCPDependency(name="github", transport="stdio")
+        result = MCPIntegrator._detect_mcp_config_drift([dep], {})
+        assert result == set()
+
+    def test_multiple_deps_mixed_drift(self):
+        """Only drifted deps are returned in a mixed set."""
+        deps = [
+            MCPDependency(name="unchanged", transport="stdio"),
+            MCPDependency(name="changed", transport="http", url="https://new.example.com"),
+        ]
+        stored = {
+            "unchanged": {"name": "unchanged", "transport": "stdio"},
+            "changed": {
+                "name": "changed", "transport": "http",
+                "url": "https://old.example.com",
+            },
+        }
+        result = MCPIntegrator._detect_mcp_config_drift(deps, stored)
+        assert result == {"changed"}
+
+    def test_no_drift_when_registry_field_matches(self):
+        """No drift when registry field (True/None) is the same."""
+        dep = MCPDependency(name="github")
+        stored = {"github": {"name": "github"}}
+        result = MCPIntegrator._detect_mcp_config_drift([dep], stored)
+        assert result == set()
+
+    def test_drift_when_headers_added(self):
+        """Drift detected when headers are added to existing server."""
+        dep = MCPDependency(
+            name="github", headers={"Authorization": "Bearer token"}
+        )
+        stored = {"github": {"name": "github"}}
+        result = MCPIntegrator._detect_mcp_config_drift([dep], stored)
+        assert result == {"github"}
+
+    def test_plain_strings_are_skipped(self):
+        """Plain string deps (no to_dict) are ignored by drift detection."""
+        result = MCPIntegrator._detect_mcp_config_drift(
+            ["ghcr.io/org/server"], {"ghcr.io/org/server": {"name": "ghcr.io/org/server"}}
+        )
+        assert result == set()
+
+
+# ---------------------------------------------------------------------------
+# get_server_configs
+# ---------------------------------------------------------------------------
+class TestGetServerConfigs:
+    """Tests for extracting server configs from MCP dependencies."""
+
+    def test_extracts_configs_from_mcp_deps(self):
+        deps = [
+            MCPDependency(name="github", transport="stdio"),
+            MCPDependency(
+                name="internal-kb", registry=False, transport="http",
+                url="https://kb.example.com",
+            ),
+        ]
+        configs = MCPIntegrator.get_server_configs(deps)
+        assert configs == {
+            "github": {"name": "github", "transport": "stdio"},
+            "internal-kb": {
+                "name": "internal-kb", "registry": False,
+                "transport": "http", "url": "https://kb.example.com",
+            },
+        }
+
+    def test_extracts_configs_from_plain_strings(self):
+        configs = MCPIntegrator.get_server_configs(["ghcr.io/org/server"])
+        assert configs == {"ghcr.io/org/server": {"name": "ghcr.io/org/server"}}
+
+    def test_empty_list(self):
+        assert MCPIntegrator.get_server_configs([]) == {}
+
+
+# ---------------------------------------------------------------------------
+# Diff-aware install — self-defined servers with config drift
+# ---------------------------------------------------------------------------
+class TestDiffAwareSelfDefinedInstall:
+
+    @patch("apm_cli.integration.mcp_integrator.MCPIntegrator._check_self_defined_servers_needing_installation")
+    @patch("apm_cli.integration.mcp_integrator.MCPIntegrator._install_for_runtime")
+    @patch("apm_cli.integration.mcp_integrator._get_console", return_value=None)
+    def test_config_drift_triggers_reinstall(
+        self, _console, mock_install_runtime, mock_check
+    ):
+        """Self-defined server with config drift should be re-installed."""
+        # Server is already configured (check returns empty)
+        mock_check.return_value = []
+        mock_install_runtime.return_value = True
+
+        dep = MCPDependency(
+            name="internal-kb", transport="http",
+            url="https://new-kb.example.com", registry=False,
+        )
+        stored_configs = {
+            "internal-kb": {
+                "name": "internal-kb", "transport": "http",
+                "url": "https://old-kb.example.com", "registry": False,
+            }
+        }
+
+        count = MCPIntegrator.install(
+            [dep], runtime="vscode",
+            stored_mcp_configs=stored_configs,
+        )
+
+        assert count == 1
+        mock_install_runtime.assert_called_once()
+
+    @patch("apm_cli.integration.mcp_integrator.MCPIntegrator._check_self_defined_servers_needing_installation")
+    @patch("apm_cli.integration.mcp_integrator.MCPIntegrator._install_for_runtime")
+    @patch("apm_cli.integration.mcp_integrator._get_console", return_value=None)
+    def test_no_drift_keeps_skip(
+        self, _console, mock_install_runtime, mock_check
+    ):
+        """Self-defined server with no config drift should still be skipped."""
+        mock_check.return_value = []
+
+        dep = MCPDependency(
+            name="internal-kb", transport="http",
+            url="https://kb.example.com", registry=False,
+        )
+        stored_configs = {
+            "internal-kb": {
+                "name": "internal-kb", "transport": "http",
+                "url": "https://kb.example.com", "registry": False,
+            }
+        }
+
+        count = MCPIntegrator.install(
+            [dep], runtime="vscode",
+            stored_mcp_configs=stored_configs,
+        )
+
+        assert count == 0
+        mock_install_runtime.assert_not_called()
+
+    @patch("apm_cli.integration.mcp_integrator.MCPIntegrator._check_self_defined_servers_needing_installation")
+    @patch("apm_cli.integration.mcp_integrator.MCPIntegrator._install_for_runtime")
+    def test_drift_shows_updated_label(
+        self, mock_install_runtime, mock_check
+    ):
+        """Config-drifted server should show 'updated' in CLI output."""
+        mock_check.return_value = []
+        mock_install_runtime.return_value = True
+        mock_console = MagicMock()
+
+        dep = MCPDependency(
+            name="internal-kb", transport="http",
+            url="https://new-kb.example.com", registry=False,
+        )
+        stored_configs = {
+            "internal-kb": {
+                "name": "internal-kb", "transport": "http",
+                "url": "https://old-kb.example.com", "registry": False,
+            }
+        }
+
+        with patch(
+            "apm_cli.integration.mcp_integrator._get_console",
+            return_value=mock_console,
+        ):
+            count = MCPIntegrator.install(
+                [dep], runtime="vscode",
+                stored_mcp_configs=stored_configs,
+            )
+
+        assert count == 1
+        printed_lines = "\n".join(
+            str(call.args[0]) for call in mock_console.print.call_args_list if call.args
+        )
+        assert "updated" in printed_lines
+
+    @patch("apm_cli.integration.mcp_integrator._rich_success")
+    @patch("apm_cli.integration.mcp_integrator.MCPIntegrator._check_self_defined_servers_needing_installation")
+    @patch("apm_cli.integration.mcp_integrator.MCPIntegrator._install_for_runtime")
+    @patch("apm_cli.integration.mcp_integrator._get_console", return_value=None)
+    def test_no_stored_configs_preserves_existing_behavior(
+        self, _console, mock_install_runtime, mock_check, mock_rich_success
+    ):
+        """Without stored configs (first install), behavior unchanged."""
+        mock_check.return_value = []
+
+        dep = MCPDependency(
+            name="internal-kb", transport="http",
+            url="https://kb.example.com", registry=False,
+        )
+
+        count = MCPIntegrator.install([dep], runtime="vscode")
+
+        assert count == 0
+        mock_install_runtime.assert_not_called()
+        mock_rich_success.assert_called_once()
+        assert "already configured" in mock_rich_success.call_args.args[0]