feat: multi-install cleanup + upgrade discoverability

Makefile

@@ -17,6 +17,7 @@ export
 	install install-dev install-core install-python install-node install-go \
 	install-aws install-kubectl install-terraform install-ansible install-docker \
 	install-brew install-rust install-uv install-% upgrade-% uninstall-% reconcile-% \
+	reconcile-all reconcile-all-dry-run \
 	build build-dist build-wheel check-dist publish-test publish-prod \
 	clean clean-build clean-test clean-pyc clean-all \
 	scripts-perms audit-auto detect-managers upgrade-managed upgrade-dry-run \

Makefile.d/user.mk

@@ -224,6 +224,12 @@ reconcile-%: scripts-perms ## Reconcile tool installation (e.g., make reconcile-
 		echo "Error: No installer found for '$*'" >&2; exit 1; \
 	fi
 
+reconcile-all: scripts-perms ## Remove duplicate installs across ALL tools (confirm each; keeps preferred)
+	@$(PYTHON) audit.py --reconcile --all --apply
+
+reconcile-all-dry-run: scripts-perms ## Preview duplicate-install cleanup across ALL tools (removes nothing)
+	@$(PYTHON) audit.py --reconcile --all
+
 # ----------------------------------------------------------------------------
 # SYSTEM MANAGEMENT
 # ----------------------------------------------------------------------------

audit.py

@@ -1286,6 +1286,158 @@
     return 0
 
 
+def _shape_reconcile_plan(tool, installs, preferred, active, protected):
+    """Build a uniform plan dict for a tool's installations.
+
+    Each installation gets a ``preferred`` boolean (matched by path) so shell
+    consumers don't have to cross-reference the separate ``preferred`` entry.
+    """
+    pref_path = preferred.path if preferred else None
+    return {
+        "tool": tool,
+        "count": len(installs),
+        "protected": protected,
+        "preferred": preferred.to_dict() if preferred else None,
+        "active": active.to_dict() if active else None,
+        "installations": [
+            {**i.to_dict(), "preferred": bool(pref_path and i.path == pref_path)}
+            for i in installs
+        ],
+    }
+
+
+def _print_reconcile_plans(plans) -> None:
+    """Human-readable rendering of reconcile plans (non-JSON mode)."""
+    multi = [p for p in plans if p["count"] > 1]
+    if not multi:
+        print("No duplicate installations detected.", file=sys.stderr)
+        return
+    for p in multi:
+        note = " [protected]" if p["protected"] else ""
+        print(f"\n{p['tool']} — {p['count']} installations{note}:", file=sys.stderr)
+        for i in p["installations"]:
+            marks = []
+            if i.get("active"):
+                marks.append("→ active")
+            if i.get("preferred"):
+                marks.append("✓ keep")
+            suffix = ("  " + "  ".join(marks)) if marks else ""
+            ver = i.get("version") or "?"
+            print(f"   • {ver}  {i['method']}  {i['path']}{suffix}", file=sys.stderr)
+
+
+def cmd_reconcile(args: argparse.Namespace) -> int:
+    """Reconcile duplicate installations of a tool.
+
+    Without --apply this only reports the plan (which install is kept vs
+    removed, which is active on PATH). With --apply it removes the non-preferred
+    installations (aggressive mode), honoring the system-tool safelist.
+
+    Supports JSON output via CLI_AUDIT_JSON=1 for consumption by guide.sh.
+    """
+    from cli_audit.catalog import ToolCatalog
+    from cli_audit.reconcile import (
+        SYSTEM_TOOL_SAFELIST,
+        bulk_reconcile,
+        detect_installations,
+        reconcile_tool,
+    )
+
+    catalog = ToolCatalog()
+
+    def _candidates(tool):
+        try:
+            return list(catalog.get(tool).to_tool().candidates)
+        except Exception:
+            return None
+
+    def _plan(tool):
+        installs = detect_installations(tool, candidates=_candidates(tool), verbose=args.verbose)
+        preferred = installs[0] if installs else None
+        active = next((i for i in installs if i.active), None)
+        return _shape_reconcile_plan(tool, installs, preferred, active, tool in SYSTEM_TOOL_SAFELIST)
+
+    apply = getattr(args, "apply", False)
+    force = getattr(args, "yes", False)
+    do_all = getattr(args, "all", False)
+
+    # --- ALL mode ---
+    # bulk_reconcile's own "all"/"conflicts" modes only look at config.tools
+    # (a tiny user-config subset), so sweep the full catalog explicitly.
+    if do_all:
+        all_tools = list(catalog.all_tools())
+        if apply:
+            result = bulk_reconcile(
+                mode="explicit", tool_names=all_tools, reconcile_mode="aggressive",
+                force=force, verbose=args.verbose,
+            )
+            # Report only tools that actually had duplicates — the explicit
+            # sweep visits every catalog entry, but --all is about conflicts.
+            conflicts = [r for r in result.results if len(r.installations) > 1]
+            if JSON_MODE:
+                print(json.dumps({
+                    "tools_checked": result.tools_checked,
+                    "conflicts_found": result.conflicts_found,
+                    "conflicts_resolved": result.conflicts_resolved,
+                    "results": [r.to_dict() for r in conflicts],
+                    "duration_seconds": result.duration_seconds,
+                }))
+            else:
+                print(result.summary(), file=sys.stderr)
+            # Fail only on real removal errors — not on protected tools (blocked)
+            # or tools the user declined (aborted), which are expected outcomes.
+            failures = [
+                r for r in conflicts
+                if not r.success and r.action_taken not in ("blocked", "aborted", "none")
+            ]
+            return 1 if failures else 0
+        # Plan-only sweep (parallel detection, no removal)
+        result = bulk_reconcile(
+            mode="explicit", tool_names=all_tools, reconcile_mode="parallel",
+            verbose=args.verbose,
+        )
+        plans = [
+            _shape_reconcile_plan(
+                r.tool, list(r.installations), r.preferred, r.active,
+                r.tool in SYSTEM_TOOL_SAFELIST,
+            )
+            for r in result.results if len(r.installations) > 1
+        ]
+        if JSON_MODE:
+            print(json.dumps({"results": plans, "total": len(plans)}))
+        else:
+            _print_reconcile_plans(plans)
+        return 0
+
+    # --- Explicit tools ---
+    tools = args.tools
+    if not tools:
+        print("reconcile: specify one or more tools, or use --all", file=sys.stderr)
+        return 2
+
+    if apply:
+        results = []
+        for tool in tools:
+            results.append(reconcile_tool(
+                tool, mode="aggressive", candidates=_candidates(tool),
+                force=force, verbose=args.verbose,
+            ))
+        if JSON_MODE:
+            print(json.dumps({"results": [r.to_dict() for r in results]}))
+        else:
+            for r in results:
+                extra = f" ({r.error_message})" if r.error_message else ""
+                print(f"{r.tool}: {r.action_taken}{extra}", file=sys.stderr)
+        return 0 if all(r.success for r in results) else 1
+
+    plans = [_plan(t) for t in tools]
+    if JSON_MODE:
+        print(json.dumps({"results": plans}))
+    else:
+        _print_reconcile_plans(plans)
+    return 0
+
+
 def main() -> int:
     """Main entry point for audit system."""
     parser = argparse.ArgumentParser(
@@ -1323,6 +1475,26 @@
         action="store_true",
         help="Show multi-version runtime status (PHP, Python, Node.js, Ruby, Go)",
     )
+    parser.add_argument(
+        "--reconcile",
+        action="store_true",
+        help="Report (or with --apply, remove) duplicate installations of a tool",
+    )
+    parser.add_argument(
+        "--all",
+        action="store_true",
+        help="With --reconcile: operate on all tools that have duplicate installs",
+    )
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="With --reconcile: actually remove non-preferred installs (default: plan only)",
+    )
+    parser.add_argument(
+        "--yes",
+        action="store_true",
+        help="With --reconcile --apply: skip confirmation prompts (honors safelist)",
+    )
     parser.add_argument(
         "--verbose", "-v",
         action="store_true",
@@ -1339,8 +1511,16 @@
     # Setup logging
     setup_logging(verbose=args.verbose)
 
+    # Reconcile-only flags are meaningless (and silently ignored) without
+    # --reconcile — fail loudly so scripting mistakes surface.
+    if not args.reconcile and (args.all or args.apply or args.yes):
+        print("--all/--apply/--yes are only valid with --reconcile", file=sys.stderr)
+        return 2
+
     # Route to appropriate command
-    if args.versions:
+    if args.reconcile:
+        return cmd_reconcile(args)
+    elif args.versions:
         return cmd_versions(args)
     elif args.update:
         # Explicit --update flag: full update of all tools

cli_audit/reconcile.py

@@ -9,9 +9,11 @@
 from __future__ import annotations
 
 import os
+import re
 import shutil
 import subprocess
 import sys
+import threading
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass
 from functools import cmp_to_key
@@ -21,7 +23,6 @@
 from .common import vlog
 from .config import Config
 from .environment import Environment
-from .installer import validate_installation
 from .upgrade import compare_versions
 
 
@@ -213,20 +214,26 @@
 
             seen_paths.add(real_path)
 
-            # Get version and validate
+            # Get THIS install's own version by running its actual binary path.
+            # validate_installation(tool_name) only ever runs the PATH-active
+            # binary, so it would report the same version for every duplicate.
+            # version_command is deliberately omitted (it runs the tool by name,
+            # i.e. the active binary) to force path-based detection.
             version = None
             valid = True
             try:
-                success, _, version_str = validate_installation(tool_name, verbose=verbose)
-                if success and version_str:
-                    version = version_str
+                from .detection import get_version_line
+                from .collectors import extract_version_number
+                meta = _catalog_meta(tool_name)
+                line = get_version_line(real_path, tool_name, meta.get("version_flag"), None)
+                if line:
+                    version = extract_version_number(line) or line.strip()
                 else:
                     valid = False
             except Exception:
                 valid = False
-                version = "unknown"
 
-            if version is None:
+            if not version:
                 version = "unknown"
 
             # Classify installation method
@@ -247,6 +254,10 @@
 
             vlog(f"  Found: {real_path} ({method}, {version})", verbose)
 
+    # Sort by preference (highest first) so installations[0] is the preferred
+    # install — callers and the guide's "keep" marker rely on this ordering.
+    installations = sort_by_preference(installations)
+
     # Cache results
     _detection_cache[cache_key] = (installations, time.time())
 
@@ -282,18 +293,38 @@
     return _classify_via_path(path)
 
 
+# Any of these characters in a path means it is not a plain absolute binary
+# path and must never be forwarded to a package-manager OS command.
+_UNSAFE_PATH_CHARS = re.compile(r"[\s;&|`$(){}<>*?!\"'\\]")
+
+
+def _is_safe_binary_path(path: str) -> bool:
+    """Validate a resolved binary path before passing it to OS commands.
+
+    Detected paths are os.path.realpath() of executables on PATH, so they are
+    always absolute and free of shell metacharacters — but validate explicitly
+    so untrusted-looking data never reaches a package-manager query.
+    """
+    return bool(path) and path.startswith("/") and not _UNSAFE_PATH_CHARS.search(path)
+
+
 def _classify_via_queries(path: str, tool_name: str, verbose: bool) -> str:
     """Classify via package manager queries."""
+    # Validate the path before any package-manager OS command: it must be a
+    # plain absolute path with no shell metacharacters.
+    if not _is_safe_binary_path(path):
+        return _classify_via_path(path)
+
     # dpkg (Debian/Ubuntu)
     if shutil.which('dpkg'):
         try:
             result = subprocess.run(
                 ['dpkg', '-S', path],
                 capture_output=True,
                 text=True,
                 timeout=2,
                 check=False,
             )
             if result.returncode == 0:
                 vlog("  Classified as apt via dpkg query", verbose)
                 return 'apt'
@@ -303,13 +334,13 @@
     # rpm (Fedora/RHEL)
     if shutil.which('rpm'):
         try:
             result = subprocess.run(
                 ['rpm', '-qf', path],
                 capture_output=True,
                 text=True,
                 timeout=2,
                 check=False,
             )
             if result.returncode == 0:
                 vlog("  Classified as dnf/rpm via rpm query", verbose)
                 return 'dnf'
@@ -328,13 +359,13 @@
             )
             if result.returncode == 0 and tool_name in result.stdout:
                 # Verify this tool is from brew
                 formula_result = subprocess.run(
                     ['brew', 'list', tool_name],
                     capture_output=True,
                     text=True,
                     timeout=2,
                     check=False,
                 )
                 if formula_result.returncode == 0 and path in formula_result.stdout:
                     vlog("  Classified as brew via brew list", verbose)
                     return 'brew'
@@ -527,6 +558,56 @@
     return sorted_installs
 
 
+_catalog_cache: dict[str, dict] = {}
+_catalog_instance = None
+_catalog_lock = threading.Lock()
+
+
+def _catalog_meta(tool_name: str) -> dict:
+    """Return cached catalog metadata for a tool: candidates + version command.
+
+    Thread-safe (bulk_reconcile detects tools in a ThreadPool). Returns an empty
+    dict for unknown tools so callers fall back to sane defaults.
+    """
+    global _catalog_instance
+    with _catalog_lock:
+        if tool_name in _catalog_cache:
+            return _catalog_cache[tool_name]
+        meta: dict = {}
+        try:
+            inst = _catalog_instance
+            if inst is None:
+                from .catalog import ToolCatalog
+                inst = ToolCatalog()
+                # Only cache a catalog that actually loaded, so a transiently
+                # empty/unavailable filesystem doesn't poison later calls.
+                if inst.all_tools():
+                    _catalog_instance = inst
+            entry = inst.get(tool_name)
+            if entry:
+                meta["candidates"] = tuple(entry.to_tool().candidates)
+                raw = getattr(entry, "_raw_data", None) or {}
+                meta["version_flag"] = raw.get("version_flag")
+                meta["version_command"] = raw.get("version_command")
+        except Exception:
+            meta = {}
+        # Only cache successful lookups — an empty result may be transient.
+        if meta:
+            _catalog_cache[tool_name] = meta
+        return meta
+
+
+def _resolve_candidates(tool_name: str) -> tuple[str, ...] | None:
+    """Look up catalog candidates for a tool (cached).
+
+    Lets reconcile detect alternate-named duplicates (e.g. ``pip``/``pip3``,
+    ``fd``/``fdfind``) even when a caller (such as bulk_reconcile) does not pass
+    candidates. Returns None if the tool is unknown, so detection falls back to
+    the bare tool name.
+    """
+    return _catalog_meta(tool_name).get("candidates")
+
+
 def reconcile_tool(
     tool_name: str,
     mode: str = "parallel",
@@ -560,6 +641,11 @@
     if env is None:
         env = detect_environment(verbose=verbose)
 
+    # Resolve catalog candidates when a caller (e.g. bulk_reconcile) omits them,
+    # so alternate-named duplicates (pip/pip3, fd/fdfind) are still detected.
+    if candidates is None:
+        candidates = _resolve_candidates(tool_name)
+
     # Detect installations
     installations = detect_installations(tool_name, candidates, verbose)