WEB-4679: declare a scan manifest so the backend can prune uninstalled tools

scripts/coding_discovery_tools/ai_tools_discovery.py

@@ -370,7 +370,7 @@ def get_device_id(self) -> str:
         """
         return self._device_id_extractor.extract_device_id()
 
-    def detect_all_tools(self, user_home: Optional[Path] = None) -> List[Dict]:
+    def detect_all_tools(self, user_home: Optional[Path] = None, failures: Optional[set] = None) -> List[Dict]:
         """
         Detect all supported AI tools.
 
@@ -401,6 +401,9 @@ def detect_all_tools(self, user_home: Optional[Path] = None) -> List[Dict]:
             except Exception as e:
                 logger.warning(f"Error detecting {detector.tool_name}: {e}")
                 report_to_sentry(e, {"phase": "detect", "tool_name": detector.tool_name}, level="warning")
+                # Detection errored: record the tool so the caller can keep it (presence unknown != uninstalled).
+                if failures is not None:
+                    failures.add(detector.tool_name)
 
         return tools
 
@@ -2802,6 +2805,11 @@ def _on_term_signal(signum, _frame) -> None:
         # Track failed reports for persistence
         failed_reports = []
 
+        # (home_user, tool_name) detected present this run; backend set-diffs it in "completed" to prune the rest.
+        scanned_manifest = set()
+        # Detector errors this run; if non-empty, no manifest is sent so the backend won't prune.
+        incomplete_reasons = []
+
         # --- Drain pending reports from previous run ---
         with time_step("drain_pending_queue", "queue"):
             pending = load_pending_reports()
@@ -2896,7 +2904,18 @@ def _on_term_signal(signum, _frame) -> None:
                 user_home = Path.home()
             logger.info(f"  Detecting tools for user: {user} (home: {user_home})")
             with time_step("detect_tools", "detect"):
-                user_tools = detector.detect_all_tools(user_home=user_home)
+                user_detect_failures = set()
+                user_tools = detector.detect_all_tools(
+                    user_home=user_home, failures=user_detect_failures
+                )
+            # Per-user presence: a detected tool stays in the manifest even if reading it later
+            # errors (a read failure isn't an uninstall).
+            for detected in user_tools:
+                scanned_manifest.add((user, detected.get('name', 'Unknown')))
+            # A detector error means presence is unknown for this user -> mark the scan incomplete
+            # so it doesn't prune (detector.tool_name is an umbrella label, not the real row name).
+            if user_detect_failures:
+                incomplete_reasons.append(f"detector error for user {user}")
 
             if user_tools:
                 logger.info(f"    Found {len(user_tools)} tool(s) for {user}:")
@@ -2959,6 +2978,11 @@ def _on_term_signal(signum, _frame) -> None:
                         user_home = Path.home()
 
                     try:
+                        # all_tools is deduped globally; only report a tool for users who actually
+                        # detected it (i.e. it's in their manifest) to avoid phantom installs.
+                        if (user_name, tool_name) not in scanned_manifest:
+                            continue
+
                         # Filter projects to only include this user's projects
                         with time_step("filter_projects", "process"):
                             tool_filtered = detector.filter_tool_projects_by_user(tool_with_projects, user_home)
@@ -2974,6 +2998,8 @@ def _on_term_signal(signum, _frame) -> None:
                                 f"{tool_filtered.get('_config_path') or tool_filtered.get('install_path')!r} "
                                 f"not owned by this user and no per-user data"
                             )
+                            # Detected globally but not owned by this user -> drop the presence entry.
+                            scanned_manifest.discard((user_name, tool_name))
                             continue
 
                         # Ownership gate (Augment surfaces): same ~/.augment-keyed
@@ -2986,6 +3012,8 @@ def _on_term_signal(signum, _frame) -> None:
                                 f"{tool_filtered.get('_config_path') or tool_filtered.get('install_path')!r} "
                                 f"not owned by this user and no per-user data"
                             )
+                            # Detected globally but not owned by this user -> drop the presence entry.
+                            scanned_manifest.discard((user_name, tool_name))
                             continue
 
                         # Detect subscription plan for Claude Code
@@ -3197,6 +3225,8 @@ def _on_term_signal(signum, _frame) -> None:
 
             except Exception as e:
                 logger.error(f"Error processing tool {tool_name}: {e}", exc_info=True)
+                # Detected tools are already in the manifest from the detection phase, so this
+                # extraction failure can't drop a live tool.
                 report_to_sentry(e, {**sentry_ctx, "phase": "process_tool", "tool_name": tool_name}, level="warning")
                 logger.info("")
 
@@ -3233,6 +3263,8 @@ def _on_term_signal(signum, _frame) -> None:
                     "os": platform.system(),
                     "tool_count": len(tools),
                     "user_count": len(all_users),
+                    "manifest_size": len(scanned_manifest),
+                    "scan_incomplete": bool(incomplete_reasons),
                     "python_version": f"{sys.version_info.major}.{sys.version_info.minor}",
                     "script_version": SCRIPT_VERSION,
                 },
@@ -3244,11 +3276,18 @@ def _on_term_signal(signum, _frame) -> None:
         except Exception as metrics_err:
             logger.debug(f"Building/sending discovery metrics failed: {metrics_err}")
 
-        # Send scan completed event AFTER all scanning
         logger.info("Sending scan completed event...")
+        # An incomplete scan sends neither manifest nor covered scope, so the backend has no
+        # partial inventory to prune from (atomic on this event — no separate signal to lose).
+        if incomplete_reasons:
+            manifest, covered = None, None
+        else:
+            manifest = [{"home_user": hu, "tool_name": tn} for hu, tn in sorted(scanned_manifest)]
+            covered = all_users
         success, _ = send_scan_event(
             args.domain, args.api_key, device_id, run_id, "completed",
-            args.app_name, sentry_context=sentry_ctx, system_user=system_user
+            args.app_name, sentry_context=sentry_ctx, system_user=system_user,
+            manifest=manifest, covered_home_users=covered,
         )
         if success:
             logger.info("✓ Scan completed event sent successfully")

scripts/coding_discovery_tools/linux/cline/cline.py

@@ -38,6 +38,11 @@ def tool_name(self) -> str:
         return "Cline"
 
     def detect(self) -> Optional[List[Dict]]:
+        # Per-user scan (user_home set by detect_tool_for_user): scope to THIS user only, else an
+        # elevated scan enumerates every home and attributes other users' extensions to the caller.
+        scoped_home = getattr(self, 'user_home', None)
+        if scoped_home is not None:
+            return self._detect_cline_for_user(Path(scoped_home)) or None
         all_results = []
         for user_home in get_linux_user_homes():
             try:

scripts/coding_discovery_tools/linux/kilocode/kilocode.py

@@ -33,6 +33,11 @@ def tool_name(self) -> str:
         return "Kilo Code"
 
     def detect(self) -> Optional[Dict]:
+        # Per-user scan (user_home set by detect_tool_for_user): scope to THIS user only, else an
+        # elevated scan enumerates every home and attributes other users' extensions to the caller.
+        scoped_home = getattr(self, 'user_home', None)
+        if scoped_home is not None:
+            return self._check_user_for_kilocode(Path(scoped_home))
         for user_home in get_linux_user_homes():
             result = self._check_user_for_kilocode(user_home)
             if result:

scripts/coding_discovery_tools/linux/roo_code/roo_code.py

@@ -39,6 +39,11 @@ def tool_name(self) -> str:
         return "Roo Code"
 
     def detect(self) -> Optional[List[Dict]]:
+        # Per-user scan (user_home set by detect_tool_for_user): scope to THIS user only, else an
+        # elevated scan enumerates every home and attributes other users' extensions to the caller.
+        scoped_home = getattr(self, 'user_home', None)
+        if scoped_home is not None:
+            return self._detect_roo_for_user(Path(scoped_home)) or None
         all_results = []
         for user_home in get_linux_user_homes():
             try:

scripts/coding_discovery_tools/macos/cline/cline.py

@@ -76,6 +76,12 @@ def detect(self) -> Optional[List[Dict]]:
             List of dicts containing tool info for each IDE with Cline installed,
             or None if not found in any IDE
         """
+        # Per-user scan (user_home set by detect_tool_for_user): scope to THIS user only. Without
+        # this, a root scan self-enumerates /Users and attributes every user's extension to the caller.
+        scoped_home = getattr(self, 'user_home', None)
+        if scoped_home is not None:
+            return self._detect_cline_for_user(Path(scoped_home)) or None
+
         all_results = []
 
         if is_running_as_root():

scripts/coding_discovery_tools/macos/kilocode/kilocode.py

@@ -50,14 +50,20 @@ def detect(self) -> Optional[Dict]:
         Returns:
             Dict containing tool info (name, version, install_path) or None if not found
         """
+        # Per-user scan (user_home set by detect_tool_for_user): scope to THIS user only. Without
+        # this, a root scan self-enumerates /Users and attributes every user's extension to the caller.
+        scoped_home = getattr(self, 'user_home', None)
+        if scoped_home is not None:
+            return self._check_user_for_kilocode(Path(scoped_home))
+
         # When running as root, scan all user directories first
         if is_running_as_root():
             user_kilocode_info = scan_user_directories(
                 lambda user_dir: self._check_user_for_kilocode(user_dir)
             )
             if user_kilocode_info:
                 return user_kilocode_info
-        
+
         # Check current user (works for both root and regular users)
         return self._check_user_for_kilocode(Path.home())
 

scripts/coding_discovery_tools/macos/roo_code/roo_code.py

@@ -78,6 +78,12 @@ def detect(self) -> Optional[List[Dict]]:
             List of dicts containing tool info for each IDE with Roo Code installed,
             or None if not found in any IDE
         """
+        # Per-user scan (user_home set by detect_tool_for_user): scope to THIS user only. Without
+        # this, a root scan self-enumerates /Users and attributes every user's extension to the caller.
+        scoped_home = getattr(self, 'user_home', None)
+        if scoped_home is not None:
+            return self._detect_roo_for_user(Path(scoped_home)) or None
+
         all_results = []
 
         if is_running_as_root():

scripts/coding_discovery_tools/utils.py

@@ -716,6 +716,8 @@ def send_scan_event(
     scan_error: Optional[Dict] = None,
     sentry_context: Optional[Dict] = None,
     system_user: Optional[str] = None,
+    manifest: Optional[List[Dict]] = None,
+    covered_home_users: Optional[List[str]] = None,
 ) -> Tuple[bool, bool]:
     """
     Send scan lifecycle event to backend (in_progress, completed, failed).
@@ -733,6 +735,10 @@ def send_scan_event(
         system_user: Optional real human user running the scan (or None). Used by
             the backend to attribute empty machines. MUST be a real human or
             None (see ``get_audit_user``), never a junk/service identity.
+        manifest: Optional [{"home_user", "tool_name"}] seen this run; sent only on
+            "completed" so the backend set-diffs it to prune the rest.
+        covered_home_users: Optional home users covered; sent only on "completed" to
+            bound the prune scope.
 
     Returns:
         Tuple of (success, retryable): success=True if sent, retryable=True if caller should queue
@@ -755,6 +761,12 @@ def send_scan_event(
     if scan_error:
         payload["scan_error"] = scan_error
 
+    if manifest is not None:
+        payload["manifest"] = manifest
+
+    if covered_home_users is not None:
+        payload["covered_home_users"] = covered_home_users
+
     return send_report_to_backend(
         backend_url,
         api_key,

scripts/coding_discovery_tools/windows/cline/cline.py

@@ -66,6 +66,12 @@ def detect(self) -> Optional[List[Dict]]:
             List of dicts containing tool info for each IDE with Cline installed,
             or None if not found in any IDE
         """
+        # Per-user scan (user_home set by detect_tool_for_user): scope to THIS user only, else an
+        # elevated scan enumerates every home and attributes other users' extensions to the caller.
+        scoped_home = getattr(self, 'user_home', None)
+        if scoped_home is not None:
+            return self._detect_cline_for_user(Path(scoped_home)) or None
+
         all_results = []
 
         if is_running_as_admin():

scripts/coding_discovery_tools/windows/kilocode/kilocode.py

@@ -67,12 +67,18 @@ def detect(self) -> Optional[Dict]:
         Returns:
             Dict containing tool info (name, version, install_path) or None if not found
         """
+        # Per-user scan (user_home set by detect_tool_for_user): scope to THIS user only, else an
+        # elevated scan enumerates every home and attributes other users' extensions to the caller.
+        scoped_home = getattr(self, 'user_home', None)
+        if scoped_home is not None:
+            return self._check_user_for_kilocode(Path(scoped_home))
+
         # When running as administrator, scan all user directories first
         if self._is_running_as_admin():
             user_kilocode_info = self._scan_user_directories()
             if user_kilocode_info:
                 return user_kilocode_info
-        
+
         # Check current user (works for both admin and regular users)
         return self._check_user_for_kilocode(Path.home())
 

scripts/coding_discovery_tools/windows/roo_code/roo_code.py

@@ -66,6 +66,12 @@ def detect(self) -> Optional[List[Dict]]:
             List of dicts containing tool info for each IDE with Roo Code installed,
             or None if not found in any IDE
         """
+        # Per-user scan (user_home set by detect_tool_for_user): scope to THIS user only, else an
+        # elevated scan enumerates every home and attributes other users' extensions to the caller.
+        scoped_home = getattr(self, 'user_home', None)
+        if scoped_home is not None:
+            return self._detect_roo_for_user(Path(scoped_home)) or None
+
         all_results = []
 
         if is_running_as_admin():