diff --git a/docs/src/content/docs/guides/dependencies.md b/docs/src/content/docs/guides/dependencies.md
index 62f1c7cc4..4198b83d0 100644
--- a/docs/src/content/docs/guides/dependencies.md
+++ b/docs/src/content/docs/guides/dependencies.md
@@ -667,6 +667,10 @@ curl -H "Authorization: token $GITHUB_CLI_PAT" https://api.github.com/user
 - Remove circular references
 - Consider merging closely related packages
 
+#### "File conflicts during installation"
+**Problem**: Local files collide with package files during `apm install`
+**Resolution**: APM skips files that exist locally and aren't managed by APM. The diagnostic summary at the end of install shows how many files were skipped. Use `--verbose` to see which files, or `--force` to overwrite.
+
 #### "File conflicts during compilation"
 **Problem**: Multiple packages or local files have same names
 **Resolution**: Local files automatically override dependency files with same names
@@ -683,7 +687,10 @@ apm deps tree
 # Preview installation without changes
 apm install --dry-run
 
-# Enable verbose logging
+# See detailed diagnostics (skipped files, errors)
+apm install --verbose
+
+# Enable verbose logging for compilation
 apm compile --verbose
 ```
 
diff --git a/docs/src/content/docs/reference/cli-commands.md b/docs/src/content/docs/reference/cli-commands.md
index fdb53a9cc..f4e849302 100644
--- a/docs/src/content/docs/reference/cli-commands.md
+++ b/docs/src/content/docs/reference/cli-commands.md
@@ -85,7 +85,7 @@ apm install [PACKAGES...] [OPTIONS]
 - `--force` - Overwrite locally-authored files on collision
 - `--dry-run` - Show what would be installed without installing
 - `--parallel-downloads INTEGER` - Max concurrent package downloads (default: 4, 0 to disable)
-- `--verbose` - Show detailed installation information
+- `--verbose` - Show individual file paths and full error details in the diagnostic summary
 - `--trust-transitive-mcp` - Trust self-defined MCP servers from transitive packages (skip re-declaration requirement)
 
 **Behavior:**
@@ -193,7 +193,19 @@ When you run `apm install`, APM automatically integrates primitives from install
 - **Control**: Disable with `apm config set auto-integrate false`
 - **Smart updates**: Only updates when package version/commit changes
 - **Hooks**: Hook `.json` files → `.github/hooks/*.json` with scripts bundled
-- **Collision detection**: Skips local files with a warning; use `--force` to overwrite
+- **Collision detection**: Skips local files that aren't managed by APM; use `--force` to overwrite
+
+**Diagnostic Summary:**
+
+After installation completes, APM prints a grouped diagnostic summary instead of inline warnings. Categories include collisions (skipped files), sub-skill overwrites, warnings, and errors.
+
+- **Normal mode**: Shows counts and actionable tips (e.g., "9 files skipped -- use `apm install --force` to overwrite")
+- **Verbose mode** (`--verbose`): Additionally lists individual file paths grouped by package, and full error details
+
+```bash
+# See exactly which files were skipped or had issues
+apm install --verbose
+```
 
 **Claude Integration (`.claude/` present):**
 
diff --git a/src/apm_cli/commands/install.py b/src/apm_cli/commands/install.py
index 217b274f0..3d1f069a0 100644
--- a/src/apm_cli/commands/install.py
+++ b/src/apm_cli/commands/install.py
@@ -8,6 +8,7 @@
 import click
 
 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
 from ._helpers import (
     _create_minimal_apm_yml,
@@ -400,6 +401,7 @@ def install(ctx, packages, runtime, exclude, only, update, dry_run, force, verbo
         if _existing_lock:
             old_mcp_servers = builtins.set(_existing_lock.mcp_servers)
 
+        apm_diagnostics = None
         if should_install_apm and apm_deps:
             if not APM_DEPS_AVAILABLE:
                 _rich_error("APM dependency system not available")
@@ -410,7 +412,7 @@ def install(ctx, packages, runtime, exclude, only, update, dry_run, force, verbo
                 # If specific packages were requested, only install those
                 # Otherwise install all from apm.yml
                 only_pkgs = builtins.list(packages) if packages else None
-                apm_count, prompt_count, agent_count = _install_apm_dependencies(
+                apm_count, prompt_count, agent_count, apm_diagnostics = _install_apm_dependencies(
                     apm_package, update, verbose, only_pkgs, force=force,
                     parallel_downloads=parallel_downloads,
                 )
@@ -462,6 +464,8 @@ def install(ctx, packages, runtime, exclude, only, update, dry_run, force, verbo
 
         # Show beautiful post-install summary
         _rich_blank_line()
+        if apm_diagnostics and apm_diagnostics.has_diagnostics:
+            apm_diagnostics.render_summary()
         if not only:
             # Load apm.yml config for summary
             apm_config = _load_apm_config()
@@ -689,6 +693,7 @@ def _collect_descendants(node, visited=None):
         command_integrator = CommandIntegrator()
         hook_integrator = HookIntegrator()
         instruction_integrator = InstructionIntegrator()
+        diagnostics = DiagnosticCollector(verbose=verbose)
         total_prompts_integrated = 0
         total_agents_integrated = 0
         total_skills_integrated = 0
@@ -946,6 +951,7 @@ def _collect_descendants(node, visited=None):
                                     prompt_integrator.integrate_package_prompts(
                                         cached_package_info, project_root,
                                         force=force, managed_files=managed_files,
+                                        diagnostics=diagnostics,
                                     )
                                 )
                                 if prompt_result.files_integrated > 0:
@@ -969,6 +975,7 @@ def _collect_descendants(node, visited=None):
                                     agent_integrator.integrate_package_agents(
                                         cached_package_info, project_root,
                                         force=force, managed_files=managed_files,
+                                        diagnostics=diagnostics,
                                     )
                                 )
                                 if agent_result.files_integrated > 0:
@@ -1012,6 +1019,7 @@ def _collect_descendants(node, visited=None):
                                     instruction_integrator.integrate_package_instructions(
                                         cached_package_info, project_root,
                                         force=force, managed_files=managed_files,
+                                        diagnostics=diagnostics,
                                     )
                                 )
                                 if instruction_result.files_integrated > 0:
@@ -1032,6 +1040,7 @@ def _collect_descendants(node, visited=None):
                                     agent_integrator.integrate_package_agents_claude(
                                         cached_package_info, project_root,
                                         force=force, managed_files=managed_files,
+                                        diagnostics=diagnostics,
                                     )
                                 )
                                 if claude_agent_result.files_integrated > 0:
@@ -1050,6 +1059,7 @@ def _collect_descendants(node, visited=None):
                                     command_integrator.integrate_package_commands(
                                         cached_package_info, project_root,
                                         force=force, managed_files=managed_files,
+                                        diagnostics=diagnostics,
                                     )
                                 )
                                 if command_result.files_integrated > 0:
@@ -1072,6 +1082,7 @@ def _collect_descendants(node, visited=None):
                                 hook_result = hook_integrator.integrate_package_hooks(
                                     cached_package_info, project_root,
                                     force=force, managed_files=managed_files,
+                                    diagnostics=diagnostics,
                                 )
                                 if hook_result.hooks_integrated > 0:
                                     total_hooks_integrated += hook_result.hooks_integrated
@@ -1084,6 +1095,7 @@ def _collect_descendants(node, visited=None):
                                 hook_result_claude = hook_integrator.integrate_package_hooks_claude(
                                     cached_package_info, project_root,
                                     force=force, managed_files=managed_files,
+                                    diagnostics=diagnostics,
                                 )
                                 if hook_result_claude.hooks_integrated > 0:
                                     total_hooks_integrated += hook_result_claude.hooks_integrated
@@ -1097,8 +1109,9 @@ def _collect_descendants(node, visited=None):
                             package_deployed_files[dep_key] = dep_deployed
                         except Exception as e:
                             # Don't fail installation if integration fails
-                            _rich_warning(
-                                f"  ⚠ Failed to integrate primitives from cached package: {e}"
+                            diagnostics.error(
+                                f"Failed to integrate primitives from cached package: {e}",
+                                package=dep_key,
                             )
 
                     continue
@@ -1195,6 +1208,7 @@ def _collect_descendants(node, visited=None):
                                 prompt_integrator.integrate_package_prompts(
                                     package_info, project_root,
                                     force=force, managed_files=managed_files,
+                                    diagnostics=diagnostics,
                                 )
                             )
                             if prompt_result.files_integrated > 0:
@@ -1218,6 +1232,7 @@ def _collect_descendants(node, visited=None):
                                 agent_integrator.integrate_package_agents(
                                     package_info, project_root,
                                     force=force, managed_files=managed_files,
+                                    diagnostics=diagnostics,
                                 )
                             )
                             if agent_result.files_integrated > 0:
@@ -1261,6 +1276,7 @@ def _collect_descendants(node, visited=None):
                                     instruction_integrator.integrate_package_instructions(
                                         package_info, project_root,
                                         force=force, managed_files=managed_files,
+                                        diagnostics=diagnostics,
                                     )
                                 )
                                 if instruction_result.files_integrated > 0:
@@ -1281,6 +1297,7 @@ def _collect_descendants(node, visited=None):
                                     agent_integrator.integrate_package_agents_claude(
                                         package_info, project_root,
                                         force=force, managed_files=managed_files,
+                                        diagnostics=diagnostics,
                                     )
                                 )
                                 if claude_agent_result.files_integrated > 0:
@@ -1299,6 +1316,7 @@ def _collect_descendants(node, visited=None):
                                     command_integrator.integrate_package_commands(
                                         package_info, project_root,
                                         force=force, managed_files=managed_files,
+                                        diagnostics=diagnostics,
                                     )
                                 )
                                 if command_result.files_integrated > 0:
@@ -1321,6 +1339,7 @@ def _collect_descendants(node, visited=None):
                                 hook_result = hook_integrator.integrate_package_hooks(
                                     package_info, project_root,
                                     force=force, managed_files=managed_files,
+                                    diagnostics=diagnostics,
                                 )
                                 if hook_result.hooks_integrated > 0:
                                     total_hooks_integrated += hook_result.hooks_integrated
@@ -1333,6 +1352,7 @@ def _collect_descendants(node, visited=None):
                                 hook_result_claude = hook_integrator.integrate_package_hooks_claude(
                                     package_info, project_root,
                                     force=force, managed_files=managed_files,
+                                    diagnostics=diagnostics,
                                 )
                                 if hook_result_claude.hooks_integrated > 0:
                                     total_hooks_integrated += hook_result_claude.hooks_integrated
@@ -1346,7 +1366,10 @@ def _collect_descendants(node, visited=None):
                             package_deployed_files[dep_ref.get_unique_key()] = dep_deployed_fresh
                         except Exception as e:
                             # Don't fail installation if integration fails
-                            _rich_warning(f"  ⚠ Failed to integrate primitives: {e}")
+                            diagnostics.error(
+                                f"Failed to integrate primitives: {e}",
+                                package=dep_ref.get_unique_key(),
+                            )
 
                 except Exception as e:
                     display_name = (
@@ -1355,7 +1378,10 @@ def _collect_descendants(node, visited=None):
                     # Remove the progress task on error
                     if "task_id" in locals():
                         progress.remove_task(task_id)
-                    _rich_error(f"❌ Failed to install {display_name}: {e}")
+                    diagnostics.error(
+                        f"Failed to install {display_name}: {e}",
+                        package=dep_ref.get_unique_key(),
+                    )
                     # Continue with other packages instead of failing completely
                     continue
 
@@ -1415,7 +1441,7 @@ def _collect_descendants(node, visited=None):
 
         _rich_success(f"Installed {installed_count} APM dependencies")
 
-        return installed_count, total_prompts_integrated, total_agents_integrated
+        return installed_count, total_prompts_integrated, total_agents_integrated, diagnostics
 
     except Exception as e:
         raise RuntimeError(f"Failed to resolve APM dependencies: {e}")
diff --git a/src/apm_cli/integration/agent_integrator.py b/src/apm_cli/integration/agent_integrator.py
index 13b867cc8..212915ecb 100644
--- a/src/apm_cli/integration/agent_integrator.py
+++ b/src/apm_cli/integration/agent_integrator.py
@@ -102,7 +102,8 @@ def copy_agent(self, source: Path, target: Path) -> int:
     
     def integrate_package_agents(self, package_info, project_root: Path,
                                    force: bool = False,
-                                   managed_files: set = None) -> IntegrationResult:
+                                   managed_files: set = None,
+                                   diagnostics=None) -> IntegrationResult:
         """Integrate all agents from a package into .github/agents/.
         
         Deploys with clean filenames. Skips user-authored files unless force=True.
@@ -154,7 +155,7 @@ def integrate_package_agents(self, package_info, project_root: Path,
             target_path = agents_dir / target_filename
             rel_path = str(target_path.relative_to(project_root))
             
-            if self.check_collision(target_path, rel_path, managed_files, force):
+            if self.check_collision(target_path, rel_path, managed_files, force, diagnostics=diagnostics):
                 files_skipped += 1
                 continue
             
@@ -168,7 +169,7 @@ def integrate_package_agents(self, package_info, project_root: Path,
                 claude_filename = self.get_target_filename_claude(source_file, package_info.package.name)
                 claude_target = claude_agents_dir / claude_filename
                 claude_rel = str(claude_target.relative_to(project_root))
-                if not self.check_collision(claude_target, claude_rel, managed_files, force):
+                if not self.check_collision(claude_target, claude_rel, managed_files, force, diagnostics=diagnostics):
                     self.copy_agent(source_file, claude_target)
                     target_paths.append(claude_target)
         
@@ -204,7 +205,8 @@ def get_target_filename_claude(self, source_file: Path, package_name: str) -> st
     
     def integrate_package_agents_claude(self, package_info, project_root: Path,
                                           force: bool = False,
-                                          managed_files: set = None) -> IntegrationResult:
+                                          managed_files: set = None,
+                                          diagnostics=None) -> IntegrationResult:
         """Integrate all agents from a package into .claude/agents/.
         
         Deploys with clean filenames. Skips user-authored files unless force=True.
@@ -246,7 +248,7 @@ def integrate_package_agents_claude(self, package_info, project_root: Path,
             target_path = agents_dir / target_filename
             rel_path = str(target_path.relative_to(project_root))
             
-            if self.check_collision(target_path, rel_path, managed_files, force):
+            if self.check_collision(target_path, rel_path, managed_files, force, diagnostics=diagnostics):
                 files_skipped += 1
                 continue
             
diff --git a/src/apm_cli/integration/base_integrator.py b/src/apm_cli/integration/base_integrator.py
index a71084018..0d98f92b8 100644
--- a/src/apm_cli/integration/base_integrator.py
+++ b/src/apm_cli/integration/base_integrator.py
@@ -1,7 +1,6 @@
 """Base integrator with shared collision detection and sync logic."""
 
 import re
-import sys
 from pathlib import Path
 from typing import Dict, List, Optional, Set
 
@@ -9,6 +8,7 @@
 
 from apm_cli.compilation.link_resolver import UnifiedLinkResolver
 from apm_cli.primitives.discovery import discover_primitives
+from apm_cli.utils.console import _rich_warning
 
 
 @dataclass
@@ -51,6 +51,7 @@ def check_collision(
         rel_path: str,
         managed_files: Optional[Set[str]],
         force: bool,
+        diagnostics=None,
     ) -> bool:
         """Return True if *target_path* is a user-authored collision.
 
@@ -60,7 +61,8 @@ def check_collision(
         3. ``rel_path`` is **not** in the managed set (→ user-authored)
         4. ``force`` is ``False``
 
-        When a collision is detected a warning is emitted to *stderr*.
+        When *diagnostics* is provided the skip is recorded there;
+        otherwise a warning is emitted via ``_rich_warning``.
 
         .. note:: Callers must pre-normalize *managed_files* with
            forward-slash separators (see ``normalize_managed_files``).
@@ -75,11 +77,13 @@ def check_collision(
         if force:
             return False
 
-        print(
-            f"\u26a0\ufe0f  Skipping {rel_path} \u2014 local file exists (not managed by APM). "
-            f"Use 'apm install --force' to overwrite.",
-            file=sys.stderr,
-        )
+        if diagnostics is not None:
+            diagnostics.skip(rel_path)
+        else:
+            _rich_warning(
+                f"Skipping {rel_path} — local file exists (not managed by APM). "
+                f"Use 'apm install --force' to overwrite."
+            )
         return True
 
     @staticmethod
diff --git a/src/apm_cli/integration/command_integrator.py b/src/apm_cli/integration/command_integrator.py
index bebcf7589..432817849 100644
--- a/src/apm_cli/integration/command_integrator.py
+++ b/src/apm_cli/integration/command_integrator.py
@@ -102,7 +102,8 @@ def integrate_command(self, source: Path, target: Path, package_info, original_p
     
     def integrate_package_commands(self, package_info, project_root: Path,
                                     force: bool = False,
-                                    managed_files: set = None) -> IntegrationResult:
+                                    managed_files: set = None,
+                                    diagnostics=None) -> IntegrationResult:
         """Integrate all prompt files from a package as Claude commands.
         
         Deploys with clean filenames. Skips user-authored files unless force=True.
@@ -137,7 +138,7 @@ def integrate_package_commands(self, package_info, project_root: Path,
             target_path = commands_dir / f"{base_name}.md"
             rel_path = str(target_path.relative_to(project_root))
             
-            if self.check_collision(target_path, rel_path, managed_files, force):
+            if self.check_collision(target_path, rel_path, managed_files, force, diagnostics=diagnostics):
                 files_skipped += 1
                 continue
             
diff --git a/src/apm_cli/integration/hook_integrator.py b/src/apm_cli/integration/hook_integrator.py
index ff660b6a8..35a07a343 100644
--- a/src/apm_cli/integration/hook_integrator.py
+++ b/src/apm_cli/integration/hook_integrator.py
@@ -254,7 +254,8 @@ def _get_package_name(self, package_info) -> str:
 
     def integrate_package_hooks(self, package_info, project_root: Path,
                                  force: bool = False,
-                                 managed_files: set = None) -> HookIntegrationResult:
+                                 managed_files: set = None,
+                                 diagnostics=None) -> HookIntegrationResult:
         """Integrate hooks from a package into .github/hooks/ (VSCode target).
 
         Deploys hook JSON files with clean filenames and copies referenced
@@ -302,7 +303,7 @@ def integrate_package_hooks(self, package_info, project_root: Path,
             target_path = hooks_dir / target_filename
             rel_path = str(target_path.relative_to(project_root))
 
-            if self.check_collision(target_path, rel_path, managed_files, force):
+            if self.check_collision(target_path, rel_path, managed_files, force, diagnostics=diagnostics):
                 continue
 
             # Write rewritten JSON
@@ -316,7 +317,7 @@ def integrate_package_hooks(self, package_info, project_root: Path,
             # Copy referenced scripts (individual file tracking)
             for source_file, target_rel in scripts:
                 target_script = project_root / target_rel
-                if self.check_collision(target_script, target_rel, managed_files, force):
+                if self.check_collision(target_script, target_rel, managed_files, force, diagnostics=diagnostics):
                     continue
                 target_script.parent.mkdir(parents=True, exist_ok=True)
                 shutil.copy2(source_file, target_script)
@@ -331,7 +332,8 @@ def integrate_package_hooks(self, package_info, project_root: Path,
 
     def integrate_package_hooks_claude(self, package_info, project_root: Path,
                                         force: bool = False,
-                                        managed_files: set = None) -> HookIntegrationResult:
+                                        managed_files: set = None,
+                                        diagnostics=None) -> HookIntegrationResult:
         """Integrate hooks from a package into .claude/settings.json (Claude target).
 
         Merges hook definitions into the Claude settings file and copies
@@ -404,7 +406,7 @@ def integrate_package_hooks_claude(self, package_info, project_root: Path,
             # Copy referenced scripts
             for source_file, target_rel in scripts:
                 target_script = project_root / target_rel
-                if self.check_collision(target_script, target_rel, managed_files, force):
+                if self.check_collision(target_script, target_rel, managed_files, force, diagnostics=diagnostics):
                     continue
                 target_script.parent.mkdir(parents=True, exist_ok=True)
                 shutil.copy2(source_file, target_script)
diff --git a/src/apm_cli/integration/instruction_integrator.py b/src/apm_cli/integration/instruction_integrator.py
index ecb597c92..80411d355 100644
--- a/src/apm_cli/integration/instruction_integrator.py
+++ b/src/apm_cli/integration/instruction_integrator.py
@@ -44,6 +44,7 @@ def integrate_package_instructions(
         project_root: Path,
         force: bool = False,
         managed_files: Optional[Set[str]] = None,
+        diagnostics=None,
     ) -> IntegrationResult:
         """Integrate all instructions from a package into .github/instructions/.
 
@@ -74,7 +75,7 @@ def integrate_package_instructions(
             target_path = instructions_dir / source_file.name
             rel_path = str(target_path.relative_to(project_root))
 
-            if self.check_collision(target_path, rel_path, managed_files, force):
+            if self.check_collision(target_path, rel_path, managed_files, force, diagnostics=diagnostics):
                 files_skipped += 1
                 continue
 
diff --git a/src/apm_cli/integration/prompt_integrator.py b/src/apm_cli/integration/prompt_integrator.py
index 419309f25..03f4b8d8d 100644
--- a/src/apm_cli/integration/prompt_integrator.py
+++ b/src/apm_cli/integration/prompt_integrator.py
@@ -67,7 +67,8 @@ def get_target_filename(self, source_file: Path, package_name: str) -> str:
     
     def integrate_package_prompts(self, package_info, project_root: Path,
                                     force: bool = False,
-                                    managed_files: set = None) -> IntegrationResult:
+                                    managed_files: set = None,
+                                    diagnostics=None) -> IntegrationResult:
         """Integrate all prompts from a package into .github/prompts/.
         
         Deploys with clean filenames. Skips files that exist locally and
@@ -111,7 +112,7 @@ def integrate_package_prompts(self, package_info, project_root: Path,
             target_path = prompts_dir / target_filename
             rel_path = str(target_path.relative_to(project_root))
             
-            if self.check_collision(target_path, rel_path, managed_files, force):
+            if self.check_collision(target_path, rel_path, managed_files, force, diagnostics=diagnostics):
                 files_skipped += 1
                 continue
             
diff --git a/src/apm_cli/integration/skill_integrator.py b/src/apm_cli/integration/skill_integrator.py
index ae451b0c7..3e11fe3b1 100644
--- a/src/apm_cli/integration/skill_integrator.py
+++ b/src/apm_cli/integration/skill_integrator.py
@@ -436,7 +436,7 @@ def find_context_files(self, package_path: Path) -> List[Path]:
         return context_files
     
     @staticmethod
-    def _promote_sub_skills(sub_skills_dir: Path, target_skills_root: Path, parent_name: str, *, warn: bool = True, owned_by: dict[str, str] | None = None) -> tuple[int, list[Path]]:
+    def _promote_sub_skills(sub_skills_dir: Path, target_skills_root: Path, parent_name: str, *, warn: bool = True, owned_by: dict[str, str] | None = None, diagnostics=None) -> tuple[int, list[Path]]:
         """Promote sub-skills from .apm/skills/ to top-level skill entries.
 
         Args:
@@ -446,6 +446,7 @@ def _promote_sub_skills(sub_skills_dir: Path, target_skills_root: Path, parent_n
             warn: Whether to emit a warning on name collisions.
             owned_by: Map of skill_name → owner_package_name from the lockfile.
                 When provided, warnings are suppressed for self-overwrites.
+            diagnostics: Optional DiagnosticCollector for deferred warning output.
 
         Returns:
             tuple[int, list[Path]]: (count of promoted sub-skills, list of deployed dir paths)
@@ -467,13 +468,20 @@ def _promote_sub_skills(sub_skills_dir: Path, target_skills_root: Path, parent_n
                 prev_owner = (owned_by or {}).get(sub_name)
                 is_self_overwrite = prev_owner is not None and prev_owner == parent_name
                 if warn and not is_self_overwrite:
-                    try:
-                        from apm_cli.utils.console import _rich_warning
-                        _rich_warning(
-                            f"Sub-skill '{sub_name}' from '{parent_name}' overwrites existing skill at {target_skills_root.name}/{sub_name}/"
+                    if diagnostics is not None:
+                        diagnostics.overwrite(
+                            path=f"{target_skills_root.name}/{sub_name}/",
+                            package=parent_name,
+                            detail=f"Sub-skill '{sub_name}' from '{parent_name}' overwrites existing skill",
                         )
-                    except ImportError:
-                        pass
+                    else:
+                        try:
+                            from apm_cli.utils.console import _rich_warning
+                            _rich_warning(
+                                f"Sub-skill '{sub_name}' from '{parent_name}' overwrites existing skill at {target_skills_root.name}/{sub_name}/"
+                            )
+                        except ImportError:
+                            pass
                 shutil.rmtree(target)
             target.mkdir(parents=True, exist_ok=True)
             shutil.copytree(sub_skill_path, target, dirs_exist_ok=True)
@@ -503,7 +511,7 @@ def _build_skill_ownership_map(project_root: Path) -> dict[str, str]:
         return owned_by
 
     def _promote_sub_skills_standalone(
-        self, package_info, project_root: Path
+        self, package_info, project_root: Path, diagnostics=None,
     ) -> tuple[int, list[Path]]:
         """Promote sub-skills from a package that is NOT itself a skill.
 
@@ -529,7 +537,7 @@ def _promote_sub_skills_standalone(
         github_skills_root.mkdir(parents=True, exist_ok=True)
         owned_by = self._build_skill_ownership_map(project_root)
         count, deployed = self._promote_sub_skills(
-            sub_skills_dir, github_skills_root, parent_name, warn=True, owned_by=owned_by
+            sub_skills_dir, github_skills_root, parent_name, warn=True, owned_by=owned_by, diagnostics=diagnostics
         )
         all_deployed = list(deployed)
 
@@ -545,7 +553,8 @@ def _promote_sub_skills_standalone(
         return count, all_deployed
 
     def _integrate_native_skill(
-        self, package_info, project_root: Path, source_skill_md: Path
+        self, package_info, project_root: Path, source_skill_md: Path,
+        diagnostics=None,
     ) -> SkillIntegrationResult:
         """Copy a native Skill (with existing SKILL.md) to .github/skills/ and optionally .claude/skills/.
         
@@ -591,14 +600,19 @@ def _integrate_native_skill(
         else:
             # Normalize the name if validation fails
             skill_name = normalize_skill_name(raw_skill_name)
-            # Log warning about name normalization (import here to avoid circular import)
-            try:
-                from apm_cli.utils.console import _rich_warning
-                _rich_warning(
-                    f"Skill name '{raw_skill_name}' normalized to '{skill_name}' ({error_msg})"
+            if diagnostics is not None:
+                diagnostics.warn(
+                    f"Skill name '{raw_skill_name}' normalized to '{skill_name}' ({error_msg})",
+                    package=raw_skill_name,
                 )
-            except ImportError:
-                pass  # CLI not available in tests
+            else:
+                try:
+                    from apm_cli.utils.console import _rich_warning
+                    _rich_warning(
+                        f"Skill name '{raw_skill_name}' normalized to '{skill_name}' ({error_msg})"
+                    )
+                except ImportError:
+                    pass  # CLI not available in tests
         
         # Primary target: .github/skills/
         github_skill_dir = project_root / ".github" / "skills" / skill_name
@@ -631,7 +645,7 @@ def _integrate_native_skill(
         sub_skills_dir = package_path / ".apm" / "skills"
         github_skills_root = project_root / ".github" / "skills"
         owned_by = self._build_skill_ownership_map(project_root)
-        sub_skills_count, sub_deployed = self._promote_sub_skills(sub_skills_dir, github_skills_root, skill_name, warn=True, owned_by=owned_by)
+        sub_skills_count, sub_deployed = self._promote_sub_skills(sub_skills_dir, github_skills_root, skill_name, warn=True, owned_by=owned_by, diagnostics=diagnostics)
         all_target_paths.extend(sub_deployed)
         
         # === T7: Copy to .claude/skills/ (secondary - compatibility) ===
@@ -663,7 +677,7 @@ def _integrate_native_skill(
             target_paths=all_target_paths
         )
 
-    def integrate_package_skill(self, package_info, project_root: Path) -> SkillIntegrationResult:
+    def integrate_package_skill(self, package_info, project_root: Path, diagnostics=None) -> SkillIntegrationResult:
         """Integrate a package's skill into .github/skills/ directory.
         
         Copies native skills (packages with SKILL.md at root) to .github/skills/
@@ -686,7 +700,7 @@ def integrate_package_skill(self, package_info, project_root: Path) -> SkillInte
             # Even non-skill packages may ship sub-skills under .apm/skills/.
             # Promote them so Copilot can discover them independently.
             sub_skills_count, sub_deployed = self._promote_sub_skills_standalone(
-                package_info, project_root
+                package_info, project_root, diagnostics=diagnostics
             )
             return SkillIntegrationResult(
                 skill_created=False,
@@ -719,12 +733,12 @@ def integrate_package_skill(self, package_info, project_root: Path) -> SkillInte
         # Check if this is a native Skill (already has SKILL.md at root)
         source_skill_md = package_path / "SKILL.md"
         if source_skill_md.exists():
-            return self._integrate_native_skill(package_info, project_root, source_skill_md)
+            return self._integrate_native_skill(package_info, project_root, source_skill_md, diagnostics=diagnostics)
         
         # No SKILL.md at root — not a skill package.
         # Still promote any sub-skills shipped under .apm/skills/.
         sub_skills_count, sub_deployed = self._promote_sub_skills_standalone(
-            package_info, project_root
+            package_info, project_root, diagnostics=diagnostics
         )
         return SkillIntegrationResult(
             skill_created=False,
diff --git a/src/apm_cli/utils/__init__.py b/src/apm_cli/utils/__init__.py
index 67bcbaf9b..dd6df7e8c 100644
--- a/src/apm_cli/utils/__init__.py
+++ b/src/apm_cli/utils/__init__.py
@@ -12,6 +12,15 @@
     STATUS_SYMBOLS
 )
 
+from .diagnostics import (
+    Diagnostic,
+    DiagnosticCollector,
+    CATEGORY_COLLISION,
+    CATEGORY_OVERWRITE,
+    CATEGORY_WARNING,
+    CATEGORY_ERROR,
+)
+
 __all__ = [
     '_rich_success',
     '_rich_error',
@@ -21,5 +30,11 @@
     '_rich_panel',
     '_create_files_table',
     '_get_console',
-    'STATUS_SYMBOLS'
+    'STATUS_SYMBOLS',
+    'Diagnostic',
+    'DiagnosticCollector',
+    'CATEGORY_COLLISION',
+    'CATEGORY_OVERWRITE',
+    'CATEGORY_WARNING',
+    'CATEGORY_ERROR',
 ]
\ No newline at end of file
diff --git a/src/apm_cli/utils/diagnostics.py b/src/apm_cli/utils/diagnostics.py
new file mode 100644
index 000000000..fcbed95aa
--- /dev/null
+++ b/src/apm_cli/utils/diagnostics.py
@@ -0,0 +1,237 @@
+"""Diagnostic collector for structured warning/error reporting.
+
+Provides a collect-then-render pattern: integrators push diagnostics
+during install (or any command), and the collector renders a clean,
+grouped summary at the end.  This replaces inline ``print()`` /
+``_rich_warning()`` calls that previously produced noisy, repetitive
+output when many packages are involved.
+"""
+
+import threading
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional
+
+from apm_cli.utils.console import (
+    _get_console,
+    _rich_echo,
+    _rich_info,
+    _rich_warning,
+)
+
+# Diagnostic categories — used as grouping keys in render_summary()
+CATEGORY_COLLISION = "collision"
+CATEGORY_OVERWRITE = "overwrite"
+CATEGORY_WARNING = "warning"
+CATEGORY_ERROR = "error"
+
+_CATEGORY_ORDER = [CATEGORY_COLLISION, CATEGORY_OVERWRITE, CATEGORY_WARNING, CATEGORY_ERROR]
+
+
+@dataclass(frozen=True)
+class Diagnostic:
+    """Single diagnostic message produced during an operation."""
+
+    message: str
+    category: str
+    package: str = ""
+    detail: str = ""
+
+
+class DiagnosticCollector:
+    """Collects diagnostics during a multi-package operation and renders
+    a grouped summary at the end.
+
+    Thread-safe: multiple integrators may push diagnostics concurrently
+    during parallel installs.
+    """
+
+    def __init__(self, verbose: bool = False) -> None:
+        self.verbose = verbose
+        self._diagnostics: List[Diagnostic] = []
+        self._lock = threading.Lock()
+
+    # ------------------------------------------------------------------
+    # Recording helpers
+    # ------------------------------------------------------------------
+
+    def skip(self, path: str, package: str = "") -> None:
+        """Record a collision skip (file exists, not managed by APM)."""
+        with self._lock:
+            self._diagnostics.append(
+                Diagnostic(
+                    message=path,
+                    category=CATEGORY_COLLISION,
+                    package=package,
+                )
+            )
+
+    def overwrite(self, path: str, package: str = "", detail: str = "") -> None:
+        """Record a sub-skill or file overwrite."""
+        with self._lock:
+            self._diagnostics.append(
+                Diagnostic(
+                    message=path,
+                    category=CATEGORY_OVERWRITE,
+                    package=package,
+                    detail=detail,
+                )
+            )
+
+    def warn(self, message: str, package: str = "", detail: str = "") -> None:
+        """Record a general warning."""
+        with self._lock:
+            self._diagnostics.append(
+                Diagnostic(
+                    message=message,
+                    category=CATEGORY_WARNING,
+                    package=package,
+                    detail=detail,
+                )
+            )
+
+    def error(self, message: str, package: str = "", detail: str = "") -> None:
+        """Record an error (download failure, integration failure, etc.)."""
+        with self._lock:
+            self._diagnostics.append(
+                Diagnostic(
+                    message=message,
+                    category=CATEGORY_ERROR,
+                    package=package,
+                    detail=detail,
+                )
+            )
+
+    # ------------------------------------------------------------------
+    # Query helpers
+    # ------------------------------------------------------------------
+
+    @property
+    def has_diagnostics(self) -> bool:
+        """Return True if any diagnostics have been recorded."""
+        return len(self._diagnostics) > 0
+
+    @property
+    def error_count(self) -> int:
+        return sum(1 for d in self._diagnostics if d.category == CATEGORY_ERROR)
+
+    def by_category(self) -> Dict[str, List[Diagnostic]]:
+        """Return diagnostics grouped by category, preserving insertion order."""
+        groups: Dict[str, List[Diagnostic]] = {}
+        for d in self._diagnostics:
+            groups.setdefault(d.category, []).append(d)
+        return groups
+
+    # ------------------------------------------------------------------
+    # Rendering
+    # ------------------------------------------------------------------
+
+    def render_summary(self) -> None:
+        """Render a grouped diagnostic summary to the console.
+
+        In normal mode, shows counts and actionable hints.
+        In verbose mode, also lists individual file paths / messages.
+        """
+        if not self._diagnostics:
+            return
+
+        groups = self.by_category()
+
+        console = _get_console()
+        # Separator line
+        if console:
+            try:
+                console.print()
+                console.print("── Diagnostics ──", style="bold cyan")
+            except Exception:
+                _rich_echo("")
+                _rich_echo("── Diagnostics ──", color="cyan", bold=True)
+        else:
+            _rich_echo("")
+            _rich_echo("── Diagnostics ──", color="cyan", bold=True)
+
+        for cat in _CATEGORY_ORDER:
+            items = groups.get(cat)
+            if not items:
+                continue
+
+            if cat == CATEGORY_COLLISION:
+                self._render_collision_group(items)
+            elif cat == CATEGORY_OVERWRITE:
+                self._render_overwrite_group(items)
+            elif cat == CATEGORY_WARNING:
+                self._render_warning_group(items)
+            elif cat == CATEGORY_ERROR:
+                self._render_error_group(items)
+
+        if console:
+            try:
+                console.print()
+            except Exception:
+                _rich_echo("")
+        else:
+            _rich_echo("")
+
+    # -- Per-category renderers ------------------------------------
+
+    def _render_collision_group(self, items: List[Diagnostic]) -> None:
+        count = len(items)
+        noun = "file" if count == 1 else "files"
+        _rich_warning(
+            f"  ⚠ {count} {noun} skipped — local files exist, not managed by APM"
+        )
+        _rich_info("    Use 'apm install --force' to overwrite")
+        if not self.verbose:
+            _rich_info("    Run with --verbose to see individual files")
+        else:
+            # Group by package for readability
+            by_pkg = _group_by_package(items)
+            for pkg, diags in by_pkg.items():
+                if pkg:
+                    _rich_echo(f"    [{pkg}]", color="dim")
+                for d in diags:
+                    _rich_echo(f"      └─ {d.message}", color="dim")
+
+    def _render_overwrite_group(self, items: List[Diagnostic]) -> None:
+        count = len(items)
+        noun = "sub-skill" if count == 1 else "sub-skills"
+        _rich_warning(
+            f"  ⚠ {count} {noun} overwrote existing skills"
+        )
+        if not self.verbose:
+            _rich_info("    Run with --verbose to see details")
+        else:
+            by_pkg = _group_by_package(items)
+            for pkg, diags in by_pkg.items():
+                if pkg:
+                    _rich_echo(f"    [{pkg}]", color="dim")
+                for d in diags:
+                    detail_str = f" — {d.detail}" if d.detail else ""
+                    _rich_echo(f"      └─ {d.message}{detail_str}", color="dim")
+
+    def _render_warning_group(self, items: List[Diagnostic]) -> None:
+        for d in items:
+            pkg_prefix = f"[{d.package}] " if d.package else ""
+            _rich_warning(f"  ⚠ {pkg_prefix}{d.message}")
+            if d.detail and self.verbose:
+                _rich_echo(f"    └─ {d.detail}", color="dim")
+
+    def _render_error_group(self, items: List[Diagnostic]) -> None:
+        count = len(items)
+        noun = "package" if count == 1 else "packages"
+        _rich_echo(f"  ✗ {count} {noun} failed:", color="red")
+        for d in items:
+            pkg_prefix = f"{d.package} — " if d.package else ""
+            _rich_echo(f"    └─ {pkg_prefix}{d.message}", color="red")
+            if d.detail and self.verbose:
+                _rich_echo(f"         {d.detail}", color="dim")
+
+
+def _group_by_package(items: List[Diagnostic]) -> Dict[str, List[Diagnostic]]:
+    """Group diagnostics by package, preserving insertion order.
+
+    Items with an empty package key are collected under ``""``.
+    """
+    groups: Dict[str, List[Diagnostic]] = {}
+    for d in items:
+        groups.setdefault(d.package, []).append(d)
+    return groups
diff --git a/tests/unit/integration/test_deployed_files_manifest.py b/tests/unit/integration/test_deployed_files_manifest.py
index a650764cd..e67522da6 100644
--- a/tests/unit/integration/test_deployed_files_manifest.py
+++ b/tests/unit/integration/test_deployed_files_manifest.py
@@ -18,6 +18,7 @@
 from pathlib import Path
 
 from apm_cli.deps.lockfile import LockedDependency, LockFile
+from apm_cli.integration.base_integrator import BaseIntegrator
 from apm_cli.integration.prompt_integrator import PromptIntegrator
 from apm_cli.integration.agent_integrator import AgentIntegrator
 from apm_cli.integration.command_integrator import CommandIntegrator
@@ -29,6 +30,7 @@
     ResolvedReference,
     GitReferenceType,
 )
+from apm_cli.utils.diagnostics import DiagnosticCollector, CATEGORY_COLLISION
 
 
 def _make_package_info(tmp_path: Path, name: str = "test-pkg",
@@ -821,7 +823,7 @@ class TestCollisionWarningOutput:
     """Verify collision detection emits warning message to stderr."""
 
     def test_prompt_collision_warns_on_stderr(self, tmp_path: Path, capsys):
-        """Prompt collision should print warning to stderr."""
+        """Prompt collision should print warning via _rich_warning."""
         prompts_dir = tmp_path / ".github" / "prompts"
         prompts_dir.mkdir(parents=True)
         (prompts_dir / "review.prompt.md").write_text("# user")
@@ -833,11 +835,12 @@ def test_prompt_collision_warns_on_stderr(self, tmp_path: Path, capsys):
             info, tmp_path, force=False, managed_files=set()
         )
         captured = capsys.readouterr()
-        assert "Skipping" in captured.err
-        assert "--force" in captured.err or "apm install --force" in captured.err
+        output = captured.out + captured.err
+        assert "Skipping" in output
+        assert "--force" in output or "apm install --force" in output
 
     def test_agent_collision_warns_on_stderr(self, tmp_path: Path, capsys):
-        """Agent collision should print warning to stderr."""
+        """Agent collision should print warning via _rich_warning."""
         agents_dir = tmp_path / ".github" / "agents"
         agents_dir.mkdir(parents=True)
         (agents_dir / "security.agent.md").write_text("# user")
@@ -849,11 +852,12 @@ def test_agent_collision_warns_on_stderr(self, tmp_path: Path, capsys):
             info, tmp_path, force=False, managed_files=set()
         )
         captured = capsys.readouterr()
-        assert "Skipping" in captured.err
-        assert "--force" in captured.err or "apm install --force" in captured.err
+        output = captured.out + captured.err
+        assert "Skipping" in output
+        assert "--force" in output or "apm install --force" in output
 
     def test_command_collision_warns_on_stderr(self, tmp_path: Path, capsys):
-        """Command collision should print warning to stderr."""
+        """Command collision should print warning via _rich_warning."""
         cmds_dir = tmp_path / ".claude" / "commands"
         cmds_dir.mkdir(parents=True)
         (cmds_dir / "review.md").write_text("# user")
@@ -865,11 +869,12 @@ def test_command_collision_warns_on_stderr(self, tmp_path: Path, capsys):
             info, tmp_path, force=False, managed_files=set()
         )
         captured = capsys.readouterr()
-        assert "Skipping" in captured.err
-        assert "--force" in captured.err or "apm install --force" in captured.err
+        output = captured.out + captured.err
+        assert "Skipping" in output
+        assert "--force" in output or "apm install --force" in output
 
     def test_hook_collision_warns_on_stderr(self, tmp_path: Path, capsys):
-        """Hook collision should print warning to stderr."""
+        """Hook collision should print warning via _rich_warning."""
         hooks_dir = tmp_path / ".github" / "hooks"
         hooks_dir.mkdir(parents=True)
         (hooks_dir / "test-pkg-hooks.json").write_text('{"user": true}')
@@ -881,8 +886,127 @@ def test_hook_collision_warns_on_stderr(self, tmp_path: Path, capsys):
             info, tmp_path, force=False, managed_files=set()
         )
         captured = capsys.readouterr()
-        assert "Skipping" in captured.err
-        assert "--force" in captured.err or "apm install --force" in captured.err
+        output = captured.out + captured.err
+        assert "Skipping" in output
+        assert "--force" in output or "apm install --force" in output
+
+
+# ---------------------------------------------------------------------------
+# 11b. check_collision() diagnostics parameter
+# ---------------------------------------------------------------------------
+
+
+class TestCheckCollisionDiagnostics:
+    """Verify check_collision() routes output to DiagnosticCollector when provided."""
+
+    def test_collision_recorded_in_diagnostics(self, tmp_path: Path):
+        """When diagnostics is provided, collision is recorded via skip()."""
+        target = tmp_path / "review.prompt.md"
+        target.write_text("# user version")
+        diag = DiagnosticCollector()
+
+        result = BaseIntegrator.check_collision(
+            target, ".github/prompts/review.prompt.md",
+            managed_files=set(), force=False, diagnostics=diag,
+        )
+
+        assert result is True
+        entries = diag.by_category().get(CATEGORY_COLLISION, [])
+        assert len(entries) == 1
+        assert entries[0].message == ".github/prompts/review.prompt.md"
+
+    def test_collision_no_stdout_when_diagnostics_provided(self, tmp_path: Path, capsys):
+        """When diagnostics is provided, nothing is printed to stdout/stderr."""
+        target = tmp_path / "agent.md"
+        target.write_text("# user")
+        diag = DiagnosticCollector()
+
+        BaseIntegrator.check_collision(
+            target, ".github/agents/agent.md",
+            managed_files=set(), force=False, diagnostics=diag,
+        )
+
+        captured = capsys.readouterr()
+        assert "Skipping" not in captured.out + captured.err
+
+    def test_fallback_warning_when_diagnostics_none(self, tmp_path: Path, capsys):
+        """When diagnostics is None, _rich_warning() fallback fires."""
+        target = tmp_path / "hook.json"
+        target.write_text("{}")
+
+        result = BaseIntegrator.check_collision(
+            target, ".github/hooks/hook.json",
+            managed_files=set(), force=False, diagnostics=None,
+        )
+
+        assert result is True
+        captured = capsys.readouterr()
+        output = captured.out + captured.err
+        assert "Skipping" in output
+        assert "--force" in output or "apm install --force" in output
+
+    def test_no_collision_no_diagnostic_recorded(self, tmp_path: Path):
+        """When there is no collision, diagnostics remains empty."""
+        target = tmp_path / "missing.md"  # does not exist on disk
+        diag = DiagnosticCollector()
+
+        result = BaseIntegrator.check_collision(
+            target, ".github/prompts/missing.md",
+            managed_files=set(), force=False, diagnostics=diag,
+        )
+
+        assert result is False
+        assert not diag.has_diagnostics
+
+    def test_force_bypasses_diagnostics(self, tmp_path: Path):
+        """force=True skips collision even when diagnostics is provided."""
+        target = tmp_path / "cmd.md"
+        target.write_text("# user")
+        diag = DiagnosticCollector()
+
+        result = BaseIntegrator.check_collision(
+            target, ".claude/commands/cmd.md",
+            managed_files=set(), force=True, diagnostics=diag,
+        )
+
+        assert result is False
+        assert not diag.has_diagnostics
+
+    def test_managed_file_bypasses_diagnostics(self, tmp_path: Path):
+        """File in managed_files is not a collision — no diagnostic recorded."""
+        target = tmp_path / "review.prompt.md"
+        target.write_text("# managed")
+        diag = DiagnosticCollector()
+
+        result = BaseIntegrator.check_collision(
+            target, ".github/prompts/review.prompt.md",
+            managed_files={".github/prompts/review.prompt.md"},
+            force=False, diagnostics=diag,
+        )
+
+        assert result is False
+        assert not diag.has_diagnostics
+
+    def test_multiple_collisions_accumulate(self, tmp_path: Path):
+        """Multiple collisions accumulate in the same collector."""
+        diag = DiagnosticCollector()
+        paths = [
+            ".github/prompts/a.prompt.md",
+            ".github/agents/b.agent.md",
+            ".claude/commands/c.md",
+        ]
+        for rel in paths:
+            target = tmp_path / Path(rel).name
+            target.write_text("# user")
+            BaseIntegrator.check_collision(
+                target, rel, managed_files=set(),
+                force=False, diagnostics=diag,
+            )
+
+        entries = diag.by_category().get(CATEGORY_COLLISION, [])
+        assert len(entries) == 3
+        recorded_paths = {e.message for e in entries}
+        assert recorded_paths == set(paths)
 
 
 # ---------------------------------------------------------------------------
diff --git a/tests/unit/test_diagnostics.py b/tests/unit/test_diagnostics.py
new file mode 100644
index 000000000..a0c2ea928
--- /dev/null
+++ b/tests/unit/test_diagnostics.py
@@ -0,0 +1,347 @@
+"""Unit tests for the DiagnosticCollector and related utilities."""
+
+import threading
+from dataclasses import FrozenInstanceError
+from unittest.mock import call, patch
+
+import pytest
+
+from apm_cli.utils.diagnostics import (
+    CATEGORY_COLLISION,
+    CATEGORY_ERROR,
+    CATEGORY_OVERWRITE,
+    CATEGORY_WARNING,
+    Diagnostic,
+    DiagnosticCollector,
+    _group_by_package,
+)
+
+# ── Diagnostic dataclass ────────────────────────────────────────────
+
+
+class TestDiagnosticDataclass:
+    def test_creation_required_fields(self):
+        d = Diagnostic(message="file.md", category=CATEGORY_WARNING)
+        assert d.message == "file.md"
+        assert d.category == CATEGORY_WARNING
+        assert d.package == ""
+        assert d.detail == ""
+
+    def test_creation_all_fields(self):
+        d = Diagnostic(
+            message="readme.md",
+            category=CATEGORY_ERROR,
+            package="my-pkg",
+            detail="download failed",
+        )
+        assert d.message == "readme.md"
+        assert d.category == CATEGORY_ERROR
+        assert d.package == "my-pkg"
+        assert d.detail == "download failed"
+
+    def test_frozen_immutable(self):
+        d = Diagnostic(message="x", category=CATEGORY_WARNING)
+        with pytest.raises(FrozenInstanceError):
+            d.message = "y"
+
+    def test_equality(self):
+        a = Diagnostic(message="f", category=CATEGORY_ERROR, package="p", detail="d")
+        b = Diagnostic(message="f", category=CATEGORY_ERROR, package="p", detail="d")
+        assert a == b
+
+    def test_inequality(self):
+        a = Diagnostic(message="f", category=CATEGORY_ERROR)
+        b = Diagnostic(message="f", category=CATEGORY_WARNING)
+        assert a != b
+
+
+# ── DiagnosticCollector — recording ─────────────────────────────────
+
+
+class TestDiagnosticCollectorRecording:
+    def test_skip_records_collision(self):
+        dc = DiagnosticCollector()
+        dc.skip("path/file.md", package="pkg-a")
+        items = dc.by_category()
+        assert CATEGORY_COLLISION in items
+        assert len(items[CATEGORY_COLLISION]) == 1
+        d = items[CATEGORY_COLLISION][0]
+        assert d.message == "path/file.md"
+        assert d.package == "pkg-a"
+
+    def test_overwrite_records_overwrite(self):
+        dc = DiagnosticCollector()
+        dc.overwrite("rules.md", package="pkg-b", detail="replaced")
+        items = dc.by_category()
+        assert CATEGORY_OVERWRITE in items
+        d = items[CATEGORY_OVERWRITE][0]
+        assert d.message == "rules.md"
+        assert d.detail == "replaced"
+
+    def test_warn_records_warning(self):
+        dc = DiagnosticCollector()
+        dc.warn("something odd", package="pkg-c", detail="extra info")
+        items = dc.by_category()
+        assert CATEGORY_WARNING in items
+        d = items[CATEGORY_WARNING][0]
+        assert d.message == "something odd"
+        assert d.package == "pkg-c"
+        assert d.detail == "extra info"
+
+    def test_error_records_error(self):
+        dc = DiagnosticCollector()
+        dc.error("download failed", package="pkg-d", detail="404")
+        items = dc.by_category()
+        assert CATEGORY_ERROR in items
+        d = items[CATEGORY_ERROR][0]
+        assert d.message == "download failed"
+        assert d.detail == "404"
+
+    def test_multiple_diagnostics_across_categories(self):
+        dc = DiagnosticCollector()
+        dc.skip("a.md", package="p1")
+        dc.overwrite("b.md", package="p2")
+        dc.warn("w", package="p3")
+        dc.error("e", package="p4")
+        groups = dc.by_category()
+        assert len(groups) == 4
+        assert len(groups[CATEGORY_COLLISION]) == 1
+        assert len(groups[CATEGORY_OVERWRITE]) == 1
+        assert len(groups[CATEGORY_WARNING]) == 1
+        assert len(groups[CATEGORY_ERROR]) == 1
+
+
+# ── DiagnosticCollector — query helpers ─────────────────────────────
+
+
+class TestDiagnosticCollectorQueryHelpers:
+    def test_has_diagnostics_false_when_empty(self):
+        dc = DiagnosticCollector()
+        assert dc.has_diagnostics is False
+
+    def test_has_diagnostics_true_after_recording(self):
+        dc = DiagnosticCollector()
+        dc.warn("w")
+        assert dc.has_diagnostics is True
+
+    def test_error_count_zero(self):
+        dc = DiagnosticCollector()
+        dc.warn("w")
+        assert dc.error_count == 0
+
+    def test_error_count_returns_correct_count(self):
+        dc = DiagnosticCollector()
+        dc.error("e1")
+        dc.error("e2")
+        dc.warn("w")
+        assert dc.error_count == 2
+
+    def test_by_category_groups_correctly(self):
+        dc = DiagnosticCollector()
+        dc.skip("s1")
+        dc.skip("s2")
+        dc.error("e1")
+        groups = dc.by_category()
+        assert len(groups[CATEGORY_COLLISION]) == 2
+        assert len(groups[CATEGORY_ERROR]) == 1
+        assert CATEGORY_WARNING not in groups
+
+    def test_by_category_preserves_insertion_order(self):
+        dc = DiagnosticCollector()
+        dc.skip("first")
+        dc.skip("second")
+        dc.skip("third")
+        collisions = dc.by_category()[CATEGORY_COLLISION]
+        assert [d.message for d in collisions] == ["first", "second", "third"]
+
+
+# ── DiagnosticCollector — rendering ─────────────────────────────────
+
+_MOCK_BASE = "apm_cli.utils.diagnostics"
+
+
+class TestDiagnosticCollectorRendering:
+    @patch(f"{_MOCK_BASE}._get_console", return_value=None)
+    @patch(f"{_MOCK_BASE}._rich_echo")
+    @patch(f"{_MOCK_BASE}._rich_warning")
+    @patch(f"{_MOCK_BASE}._rich_info")
+    def test_render_summary_does_nothing_when_empty(
+        self, mock_info, mock_warning, mock_echo, mock_console
+    ):
+        dc = DiagnosticCollector()
+        dc.render_summary()
+        mock_echo.assert_not_called()
+        mock_warning.assert_not_called()
+        mock_info.assert_not_called()
+
+    @patch(f"{_MOCK_BASE}._get_console", return_value=None)
+    @patch(f"{_MOCK_BASE}._rich_echo")
+    @patch(f"{_MOCK_BASE}._rich_warning")
+    @patch(f"{_MOCK_BASE}._rich_info")
+    def test_render_summary_normal_shows_counts_not_files(
+        self, mock_info, mock_warning, mock_echo, mock_console
+    ):
+        dc = DiagnosticCollector(verbose=False)
+        dc.skip("a.md", package="p1")
+        dc.skip("b.md", package="p1")
+        dc.render_summary()
+        # Should mention count
+        warning_texts = [str(c) for c in mock_warning.call_args_list]
+        assert any("2 files skipped" in t for t in warning_texts)
+        # Should NOT list individual file paths
+        echo_texts = [str(c) for c in mock_echo.call_args_list]
+        assert not any("a.md" in t for t in echo_texts)
+        assert not any("b.md" in t for t in echo_texts)
+
+    @patch(f"{_MOCK_BASE}._get_console", return_value=None)
+    @patch(f"{_MOCK_BASE}._rich_echo")
+    @patch(f"{_MOCK_BASE}._rich_warning")
+    @patch(f"{_MOCK_BASE}._rich_info")
+    def test_render_summary_verbose_shows_file_paths(
+        self, mock_info, mock_warning, mock_echo, mock_console
+    ):
+        dc = DiagnosticCollector(verbose=True)
+        dc.skip("a.md", package="p1")
+        dc.render_summary()
+        echo_texts = [str(c) for c in mock_echo.call_args_list]
+        assert any("a.md" in t for t in echo_texts)
+
+    @patch(f"{_MOCK_BASE}._get_console", return_value=None)
+    @patch(f"{_MOCK_BASE}._rich_echo")
+    @patch(f"{_MOCK_BASE}._rich_warning")
+    @patch(f"{_MOCK_BASE}._rich_info")
+    def test_collision_group_shows_force_hint(
+        self, mock_info, mock_warning, mock_echo, mock_console
+    ):
+        dc = DiagnosticCollector()
+        dc.skip("f.md")
+        dc.render_summary()
+        info_texts = [str(c) for c in mock_info.call_args_list]
+        assert any("--force" in t for t in info_texts)
+
+    @patch(f"{_MOCK_BASE}._get_console", return_value=None)
+    @patch(f"{_MOCK_BASE}._rich_echo")
+    @patch(f"{_MOCK_BASE}._rich_warning")
+    @patch(f"{_MOCK_BASE}._rich_info")
+    def test_overwrite_group_shows_overwrote_message(
+        self, mock_info, mock_warning, mock_echo, mock_console
+    ):
+        dc = DiagnosticCollector()
+        dc.overwrite("skill.md", package="pkg")
+        dc.render_summary()
+        warning_texts = [str(c) for c in mock_warning.call_args_list]
+        assert any("sub-skill" in t and "overwrote" in t for t in warning_texts)
+
+    @patch(f"{_MOCK_BASE}._get_console", return_value=None)
+    @patch(f"{_MOCK_BASE}._rich_echo")
+    @patch(f"{_MOCK_BASE}._rich_warning")
+    @patch(f"{_MOCK_BASE}._rich_info")
+    def test_error_group_shows_packages_failed(
+        self, mock_info, mock_warning, mock_echo, mock_console
+    ):
+        dc = DiagnosticCollector()
+        dc.error("timeout", package="pkg-x")
+        dc.render_summary()
+        echo_texts = [str(c) for c in mock_echo.call_args_list]
+        assert any("failed" in t for t in echo_texts)
+        assert any("pkg-x" in t for t in echo_texts)
+
+    @patch(f"{_MOCK_BASE}._get_console", return_value=None)
+    @patch(f"{_MOCK_BASE}._rich_echo")
+    @patch(f"{_MOCK_BASE}._rich_warning")
+    @patch(f"{_MOCK_BASE}._rich_info")
+    def test_warning_group_shows_individual_warnings(
+        self, mock_info, mock_warning, mock_echo, mock_console
+    ):
+        dc = DiagnosticCollector()
+        dc.warn("something weird", package="pkg-w")
+        dc.render_summary()
+        warning_texts = [str(c) for c in mock_warning.call_args_list]
+        assert any("something weird" in t for t in warning_texts)
+        assert any("pkg-w" in t for t in warning_texts)
+
+    @patch(f"{_MOCK_BASE}._get_console", return_value=None)
+    @patch(f"{_MOCK_BASE}._rich_echo")
+    @patch(f"{_MOCK_BASE}._rich_warning")
+    @patch(f"{_MOCK_BASE}._rich_info")
+    def test_render_summary_handles_all_categories(
+        self, mock_info, mock_warning, mock_echo, mock_console
+    ):
+        dc = DiagnosticCollector(verbose=True)
+        dc.skip("collision.md", package="p1")
+        dc.overwrite("over.md", package="p2", detail="replaced")
+        dc.warn("watch out", package="p3")
+        dc.error("boom", package="p4", detail="stack trace")
+        dc.render_summary()
+
+        all_texts = (
+            [str(c) for c in mock_echo.call_args_list]
+            + [str(c) for c in mock_warning.call_args_list]
+            + [str(c) for c in mock_info.call_args_list]
+        )
+        combined = " ".join(all_texts)
+        # All categories should appear
+        assert "skipped" in combined
+        assert "overwrote" in combined
+        assert "watch out" in combined
+        assert "failed" in combined
+
+
+# ── Thread safety ───────────────────────────────────────────────────
+
+
+class TestDiagnosticCollectorThreadSafety:
+    def test_concurrent_skip_calls_preserve_all_data(self):
+        dc = DiagnosticCollector()
+        num_threads = 10
+        items_per_thread = 100
+        barrier = threading.Barrier(num_threads)
+
+        def worker(tid: int):
+            barrier.wait()
+            for i in range(items_per_thread):
+                dc.skip(f"t{tid}-{i}.md", package=f"pkg-{tid}")
+
+        threads = [threading.Thread(target=worker, args=(t,)) for t in range(num_threads)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        total = num_threads * items_per_thread
+        assert len(dc.by_category()[CATEGORY_COLLISION]) == total
+
+
+# ── _group_by_package helper ────────────────────────────────────────
+
+
+class TestGroupByPackage:
+    def test_groups_by_package(self):
+        items = [
+            Diagnostic(message="a", category=CATEGORY_WARNING, package="p1"),
+            Diagnostic(message="b", category=CATEGORY_WARNING, package="p2"),
+            Diagnostic(message="c", category=CATEGORY_WARNING, package="p1"),
+        ]
+        groups = _group_by_package(items)
+        assert list(groups.keys()) == ["p1", "p2"]
+        assert len(groups["p1"]) == 2
+        assert len(groups["p2"]) == 1
+
+    def test_empty_package_key(self):
+        items = [
+            Diagnostic(message="x", category=CATEGORY_WARNING, package=""),
+            Diagnostic(message="y", category=CATEGORY_WARNING, package="pkg"),
+        ]
+        groups = _group_by_package(items)
+        assert "" in groups
+        assert len(groups[""]) == 1
+        assert len(groups["pkg"]) == 1
+
+    def test_preserves_insertion_order(self):
+        items = [
+            Diagnostic(message="a", category=CATEGORY_WARNING, package="z"),
+            Diagnostic(message="b", category=CATEGORY_WARNING, package="a"),
+            Diagnostic(message="c", category=CATEGORY_WARNING, package="m"),
+        ]
+        groups = _group_by_package(items)
+        assert list(groups.keys()) == ["z", "a", "m"]
diff --git a/tests/unit/test_install_command.py b/tests/unit/test_install_command.py
index 6c57259dc..6eb1231cd 100644
--- a/tests/unit/test_install_command.py
+++ b/tests/unit/test_install_command.py
@@ -66,7 +66,7 @@ def test_install_no_apm_yml_with_packages_creates_minimal_apm_yml(
             mock_apm_package.from_apm_yml.return_value = mock_pkg_instance
 
             # Mock the install function to avoid actual installation
-            mock_install_apm.return_value = (0, 0, 0)  # Return tuple (installed_count, prompts_integrated, agents_integrated)
+            mock_install_apm.return_value = (0, 0, 0, MagicMock(has_diagnostics=False))
 
             result = self.runner.invoke(cli, ["install", "test/package"])
 
@@ -106,7 +106,7 @@ def test_install_no_apm_yml_with_multiple_packages(
             mock_pkg_instance.get_mcp_dependencies.return_value = []
             mock_apm_package.from_apm_yml.return_value = mock_pkg_instance
 
-            mock_install_apm.return_value = (0, 0, 0)  # Return tuple (installed_count, prompts_integrated, agents_integrated)
+            mock_install_apm.return_value = (0, 0, 0, MagicMock(has_diagnostics=False))
 
             result = self.runner.invoke(cli, ["install", "org1/pkg1", "org2/pkg2"])
 
@@ -148,7 +148,7 @@ def test_install_existing_apm_yml_preserves_behavior(
             mock_pkg_instance.get_mcp_dependencies.return_value = []
             mock_apm_package.from_apm_yml.return_value = mock_pkg_instance
 
-            mock_install_apm.return_value = (0, 0, 0)  # Return tuple (installed_count, prompts_integrated, agents_integrated)
+            mock_install_apm.return_value = (0, 0, 0, MagicMock(has_diagnostics=False))
 
             result = self.runner.invoke(cli, ["install"])
 
@@ -186,7 +186,7 @@ def test_install_auto_created_apm_yml_has_correct_metadata(
             mock_pkg_instance.get_mcp_dependencies.return_value = []
             mock_apm_package.from_apm_yml.return_value = mock_pkg_instance
 
-            mock_install_apm.return_value = (0, 0, 0)  # Return tuple (installed_count, prompts_integrated, agents_integrated)
+            mock_install_apm.return_value = (0, 0, 0, MagicMock(has_diagnostics=False))
 
             result = self.runner.invoke(cli, ["install", "test/package"])
 
