diff --git a/scripts/coding_discovery_tools/ai_tools_discovery.py b/scripts/coding_discovery_tools/ai_tools_discovery.py index a9a0c94..d6fb4e1 100644 --- a/scripts/coding_discovery_tools/ai_tools_discovery.py +++ b/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") diff --git a/scripts/coding_discovery_tools/linux/cline/cline.py b/scripts/coding_discovery_tools/linux/cline/cline.py index 2a7da94..83a8d0d 100644 --- a/scripts/coding_discovery_tools/linux/cline/cline.py +++ b/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: diff --git a/scripts/coding_discovery_tools/linux/kilocode/kilocode.py b/scripts/coding_discovery_tools/linux/kilocode/kilocode.py index 075d50d..3756ecb 100644 --- a/scripts/coding_discovery_tools/linux/kilocode/kilocode.py +++ b/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: diff --git a/scripts/coding_discovery_tools/linux/roo_code/roo_code.py b/scripts/coding_discovery_tools/linux/roo_code/roo_code.py index 7f5f2cb..2e712ac 100644 --- a/scripts/coding_discovery_tools/linux/roo_code/roo_code.py +++ b/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: diff --git a/scripts/coding_discovery_tools/macos/cline/cline.py b/scripts/coding_discovery_tools/macos/cline/cline.py index d965585..a0adfaa 100644 --- a/scripts/coding_discovery_tools/macos/cline/cline.py +++ b/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(): diff --git a/scripts/coding_discovery_tools/macos/kilocode/kilocode.py b/scripts/coding_discovery_tools/macos/kilocode/kilocode.py index 196b7d9..ae93c4f 100644 --- a/scripts/coding_discovery_tools/macos/kilocode/kilocode.py +++ b/scripts/coding_discovery_tools/macos/kilocode/kilocode.py @@ -50,6 +50,12 @@ 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( @@ -57,7 +63,7 @@ def detect(self) -> Optional[Dict]: ) 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()) diff --git a/scripts/coding_discovery_tools/macos/roo_code/roo_code.py b/scripts/coding_discovery_tools/macos/roo_code/roo_code.py index 29638eb..f8fa7d1 100644 --- a/scripts/coding_discovery_tools/macos/roo_code/roo_code.py +++ b/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(): diff --git a/scripts/coding_discovery_tools/utils.py b/scripts/coding_discovery_tools/utils.py index 4ae1878..06d0993 100644 --- a/scripts/coding_discovery_tools/utils.py +++ b/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, diff --git a/scripts/coding_discovery_tools/windows/cline/cline.py b/scripts/coding_discovery_tools/windows/cline/cline.py index 1c987a2..3900806 100644 --- a/scripts/coding_discovery_tools/windows/cline/cline.py +++ b/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(): diff --git a/scripts/coding_discovery_tools/windows/kilocode/kilocode.py b/scripts/coding_discovery_tools/windows/kilocode/kilocode.py index c5a5205..416b0bb 100644 --- a/scripts/coding_discovery_tools/windows/kilocode/kilocode.py +++ b/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()) diff --git a/scripts/coding_discovery_tools/windows/roo_code/roo_code.py b/scripts/coding_discovery_tools/windows/roo_code/roo_code.py index f0cb83a..097ab96 100644 --- a/scripts/coding_discovery_tools/windows/roo_code/roo_code.py +++ b/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(): diff --git a/tests/test_scan_completed_manifest.py b/tests/test_scan_completed_manifest.py new file mode 100644 index 0000000..be5b467 --- /dev/null +++ b/tests/test_scan_completed_manifest.py @@ -0,0 +1,510 @@ +"""WEB-4679: the "completed" scan event carries a manifest of detected (home_user, tool_name) +pairs + the covered home users, so the backend can set-diff and prune what's gone. + +Properties: the manifest is built from per-user DETECTION (not extraction success), so a read +error keeps a detected tool; only users who detected a tool get an entry (no phantom ownership); +a DETECTOR error sends no manifest (backend then skips pruning). + +Seams: TestSendScanEventManifest (send_scan_event vs a localhost server), TestCompletedEventManifestCLI +(main() via subprocess), TestManifestFromPresence (main() in-process with a mocked detector), +TestJetBrainsNamingDeterminism (prune-key naming). Only HTTP/HOME/_SENTRY_DSN/discovery_cache are mocked. +""" + +import json +import os +import subprocess +import sys +import tempfile +import threading +import unittest +from http.server import HTTPServer, BaseHTTPRequestHandler +from pathlib import Path +from unittest.mock import Mock, patch + +import scripts.coding_discovery_tools.utils as utils_mod +from scripts.coding_discovery_tools.utils import send_scan_event +from scripts.coding_discovery_tools.macos.jetbrains.jetbrains import MacOSJetBrainsDetector + +REPO_ROOT = Path(__file__).resolve().parent.parent + + +class _RecordingHandler(BaseHTTPRequestHandler): + """Records every POST body (parsed as JSON) and returns 200.""" + + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(length) + try: + self.server.requests.append(json.loads(body)) + except ValueError: + self.server.requests.append({"_raw": body.decode("utf-8", "replace")}) + + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"ok": true}') + + def log_message(self, format, *args): + pass # suppress server logs + + +class _ServerTestCase(unittest.TestCase): + """Spins up one recording HTTP server on localhost for the whole class.""" + + @classmethod + def setUpClass(cls): + cls.server = HTTPServer(("127.0.0.1", 0), _RecordingHandler) + cls.server.requests = [] + cls.port = cls.server.server_address[1] + cls.base_url = f"http://127.0.0.1:{cls.port}" + cls.thread = threading.Thread(target=cls.server.serve_forever) + cls.thread.daemon = True + cls.thread.start() + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + cls.thread.join(timeout=5) + + def setUp(self): + self.server.requests.clear() + + +class TestSendScanEventManifest(_ServerTestCase): + """utils.send_scan_event seam: manifest + covered_home_users are inserted + into the POST body only when provided (backward compatible).""" + + @patch("time.sleep") + @patch.object(utils_mod, "_SENTRY_DSN", "") + def test_completed_event_carries_manifest_and_covered_users(self, _sleep): + manifest = [{"home_user": "alice", "tool_name": "Cursor"}] + covered = ["alice", "bob"] + + success, _retryable = send_scan_event( + self.base_url, + "test-key", + "DEV-1", + "run-1", + "completed", + manifest=manifest, + covered_home_users=covered, + ) + + self.assertTrue(success) + self.assertEqual(len(self.server.requests), 1) + body = self.server.requests[0] + # Exact passthrough of both new fields. + self.assertEqual(body["scan_event"], "completed") + self.assertEqual(body["manifest"], manifest) + self.assertEqual(body["covered_home_users"], covered) + + @patch("time.sleep") + @patch.object(utils_mod, "_SENTRY_DSN", "") + def test_legacy_call_omits_both_keys(self, _sleep): + # No manifest / covered_home_users supplied -> neither key may appear + # in the payload (backward compatibility with the old call sites). + success, _retryable = send_scan_event( + self.base_url, "test-key", "DEV-1", "run-1", "in_progress" + ) + + self.assertTrue(success) + self.assertEqual(len(self.server.requests), 1) + body = self.server.requests[0] + self.assertNotIn("manifest", body) + self.assertNotIn("covered_home_users", body) + + @patch("time.sleep") + @patch.object(utils_mod, "_SENTRY_DSN", "") + def test_empty_manifest_still_sent(self, _sleep): + # Empty manifest != "no manifest": it means "zero tools in scope" and must be sent + # (key present), since the backend guard is `is not None`, not truthiness. + success, _retryable = send_scan_event( + self.base_url, + "test-key", + "DEV-1", + "run-1", + "completed", + manifest=[], + covered_home_users=["alice"], + ) + + self.assertTrue(success) + body = self.server.requests[0] + self.assertIn("manifest", body) + self.assertEqual(body["manifest"], []) + self.assertEqual(body["covered_home_users"], ["alice"]) + + +class TestCompletedEventManifestCLI(_ServerTestCase): + """End-to-end via main() subprocess: the completed event carries a + well-formed manifest + covered_home_users; lifecycle events that are not + "completed" carry neither.""" + + def _run_cli(self, timeout=600): + env = os.environ.copy() + # Throwaway HOME: isolated lock/cache so the run isn't blocked by a live lock and starts cold. + env["HOME"] = tempfile.mkdtemp(prefix="web4679_home_") + return subprocess.run( + [ + sys.executable, + "scripts/coding_discovery_tools/ai_tools_discovery.py", + "--api-key", + "test-key-000000", + "--domain", + self.base_url, + ], + cwd=str(REPO_ROOT), + capture_output=True, + text=True, + timeout=timeout, + env=env, + ) + + def test_completed_event_has_manifest_and_covered_users(self): + result = self._run_cli() + self.assertEqual(result.returncode, 0, f"stderr: {result.stderr[-2000:]}") + + completed = [ + r for r in self.server.requests if r.get("scan_event") == "completed" + ] + self.assertEqual(len(completed), 1, "expected exactly one completed event") + body = completed[0] + + # manifest: list of {home_user, tool_name} objects. + self.assertIn("manifest", body) + self.assertIsInstance(body["manifest"], list) + for entry in body["manifest"]: + self.assertIsInstance(entry, dict) + self.assertIn("home_user", entry) + self.assertIn("tool_name", entry) + self.assertIsInstance(entry["home_user"], str) + self.assertIsInstance(entry["tool_name"], str) + + # covered_home_users: list of user names (strings). + self.assertIn("covered_home_users", body) + self.assertIsInstance(body["covered_home_users"], list) + for user in body["covered_home_users"]: + self.assertIsInstance(user, str) + + def test_non_completed_events_have_no_manifest(self): + result = self._run_cli() + self.assertEqual(result.returncode, 0, f"stderr: {result.stderr[-2000:]}") + + # An in_progress event is always sent before scanning. + non_completed = [ + r + for r in self.server.requests + if r.get("scan_event") in ("in_progress", "failed") + ] + self.assertGreaterEqual( + len(non_completed), 1, "expected at least an in_progress event" + ) + for body in non_completed: + self.assertNotIn( + "manifest", body, f"{body.get('scan_event')} must not carry a manifest" + ) + self.assertNotIn( + "covered_home_users", + body, + f"{body.get('scan_event')} must not carry covered_home_users", + ) + + def test_covered_home_users_matches_full_enumeration_not_manifest(self): + # covered_home_users comes from the full enumeration, not the manifest's users — so it must + # be a superset of the manifest's user set (asserted without forcing a zero-tool user). + result = self._run_cli() + self.assertEqual(result.returncode, 0, f"stderr: {result.stderr[-2000:]}") + + completed = [ + r for r in self.server.requests if r.get("scan_event") == "completed" + ] + self.assertEqual(len(completed), 1) + body = completed[0] + + covered = set(body["covered_home_users"]) + manifest_users = {e["home_user"] for e in body["manifest"]} + # The enumerated set must cover every user that yielded a manifest entry. + self.assertTrue( + manifest_users.issubset(covered), + f"manifest users {manifest_users} not all in covered {covered}", + ) + + +class TestManifestFromPresence(unittest.TestCase): + """Manifest is built from per-user DETECTION: a read error keeps a detected tool; only users who + detected a tool get an entry; a DETECTOR error sends no manifest. Driven via main() in-process + with a mocked detector + captured send_scan_event.""" + + def setUp(self): + import scripts.coding_discovery_tools.ai_tools_discovery as adm + + self.adm = adm + self.argv = [ + "ai_tools_discovery.py", + "--api-key", + "k", + "--domain", + "http://127.0.0.1:1", + ] + + @staticmethod + def _make_tool(name): + # Distinct install_path per tool so the (name:path) dedup keeps all three. + return {"name": name, "version": "1.0", "install_path": f"/opt/{name}", "projects": []} + + def _run_main_capture_manifest(self, send_report_result=(True, False), filter_error=None, detector_failure=None): + """Run main() with three detected tools for one user: ToolOK (send), ToolHashMatch (dedup + skip), ToolErr (filter raises). detector_failure, if set, makes detect_all_tools report a + detector error. Returns the captured (manifest, covered_home_users) from the completed event.""" + adm = self.adm + + tool_ok = self._make_tool("ToolOK") + tool_hm = self._make_tool("ToolHashMatch") + tool_err = self._make_tool("ToolErr") + + detector = Mock() + detector.get_device_id.return_value = "dev-xyz" + + def _detect_all(user_home=None, failures=None): + # A detector error surfaces via the `failures` set (-> scan marked incomplete). + if detector_failure and failures is not None: + failures.add(detector_failure) + return [tool_ok, tool_hm, tool_err] + detector.detect_all_tools.side_effect = _detect_all + detector._set_canonical_vscode_copilot.return_value = None + detector.process_single_tool.side_effect = lambda t: t + + def _filter(tool_with_projects, _user_home): + if tool_with_projects["name"] == "ToolErr": + raise (filter_error if filter_error is not None else PermissionError("simulated read failure")) + return tool_with_projects + + detector.filter_tool_projects_by_user.side_effect = _filter + detector.generate_single_tool_report.side_effect = ( + lambda tool, device_id, home_user, system_user=None, run_id=None: { + "tools": [tool] + } + ) + + # Hash is derived from the tool name; the cache "matches" only for + # ToolHashMatch, forcing ToolOK down the send path and ToolHashMatch + # down the dedup short-circuit. + def _hash(tool_dict): + return "hash-" + tool_dict["name"] + + def _cached(tool_name, _user_name): + return "hash-ToolHashMatch" if tool_name == "ToolHashMatch" else None + + dc = Mock() + dc.acquire_lock.return_value = "acquired" + dc.heartbeat_start.return_value = Mock() + dc.get_cached_hash.side_effect = _cached + dc.update_tool.return_value = None + dc.UNBOUND_DIR = "/tmp/unbound-test" + dc.last_lock_error = None + + captured = {} + + def _send_scan_event(domain, api_key, device_id, run_id, scan_event, app_name=None, **kw): + if scan_event == "completed": + captured["manifest"] = kw.get("manifest") + captured["covered_home_users"] = kw.get("covered_home_users") + elif scan_event == "failed": + captured.setdefault("failed_events", []).append(kw.get("scan_error")) + return (True, None) + + with patch.object(adm.platform, "system", return_value="Darwin"), \ + patch.object(adm, "AIToolsDetector", return_value=detector), \ + patch.object(adm, "discovery_cache", dc), \ + patch.object(adm, "get_all_users_macos", return_value=["alice"]), \ + patch.object(adm, "compute_payload_hash", side_effect=_hash), \ + patch.object(adm, "send_report_to_backend", return_value=send_report_result), \ + patch.object(adm, "send_scan_event", side_effect=_send_scan_event), \ + patch.object(adm, "send_discovery_metrics", Mock()), \ + patch.object(adm, "load_pending_reports", return_value=[]), \ + patch.object(adm, "save_failed_reports", Mock()), \ + patch.object(adm, "report_to_sentry", Mock()), \ + patch.object(utils_mod, "_SENTRY_DSN", ""), \ + patch.object(sys, "argv", self.argv): + try: + adm.main() + except SystemExit: + pass + + return captured + + def test_read_error_keeps_tool_in_manifest(self): + # A tool whose config read ERRORS is still detected present -> stays in the manifest (a read failure isn't an uninstall). + captured = self._run_main_capture_manifest() + + self.assertIn("manifest", captured, "completed event was never sent") + self.assertIsNotNone(captured["manifest"], "a read error must NOT fail-close the manifest to None") + pairs = {(e["home_user"], e["tool_name"]) for e in captured["manifest"]} + + self.assertIn(("alice", "ToolOK"), pairs) # sent path + self.assertIn(("alice", "ToolHashMatch"), pairs) # hash-match (unchanged, still installed) + self.assertIn(("alice", "ToolErr"), pairs) # read errored but DETECTED -> kept + self.assertEqual(len(captured["manifest"]), 3) + + def test_upload_failure_keeps_tool_in_manifest(self): + # Presence is recorded before extraction, so a transient UPLOAD failure still keeps the tool in the manifest. + captured = self._run_main_capture_manifest(send_report_result=(False, True)) + + self.assertIn("manifest", captured, "completed event was never sent") + pairs = {(e["home_user"], e["tool_name"]) for e in captured["manifest"]} + # All three detected tools present, regardless of upload outcome / read error. + self.assertEqual( + pairs, + {("alice", "ToolOK"), ("alice", "ToolHashMatch"), ("alice", "ToolErr")}, + ) + + def test_covered_home_users_includes_user_with_no_manifest_entry(self): + # covered_home_users comes from the full enumeration, not the manifest, so a user whose + # tools all errored is still covered (bounds the prune scope correctly). + captured = self._run_main_capture_manifest() + self.assertEqual(captured.get("covered_home_users"), ["alice"]) + + def test_generic_read_error_does_not_fail_close(self): + # Regression: a generic read error used to fail-close the manifest to None (blocking all pruning); it must no longer. + captured = self._run_main_capture_manifest( + filter_error=RuntimeError("simulated generic read failure") + ) + self.assertIn("manifest", captured, "completed event was never sent") + self.assertIsNotNone( + captured["manifest"], + "a generic read error must NOT fail-close the manifest to None", + ) + pairs = {(e["home_user"], e["tool_name"]) for e in captured["manifest"]} + self.assertIn(("alice", "ToolErr"), pairs) + self.assertEqual(len(captured["manifest"]), 3) + self.assertEqual(captured.get("covered_home_users"), ["alice"]) + + def test_detector_error_sends_no_manifest(self): + # A detector error means presence is unknown this run, so NO manifest AND no covered scope + # are sent — the backend then has no partial inventory/scope to prune from. + captured = self._run_main_capture_manifest(detector_failure="ToolGhost") + self.assertIsNone(captured["manifest"], "detector error must send no manifest") + self.assertIsNone(captured.get("covered_home_users"), "no covered scope without an inventory") + + def test_per_user_detection_no_phantom_ownership(self): + # Phantom-ownership regression: all_tools is deduped globally, so a user-scoped tool one + # user has must NOT be attributed to a co-resident user who did not detect it. Alice has + # ToolA, Bob has ToolB; the manifest must contain exactly each user's own tool. + adm = self.adm + tool_a = self._make_tool("ToolA") + tool_b = self._make_tool("ToolB") + + detector = Mock() + detector.get_device_id.return_value = "dev-xyz" + + def _detect_all(user_home=None, failures=None): + home = str(user_home or "") + if home.endswith("alice"): + return [tool_a] + if home.endswith("bob"): + return [tool_b] + return [] + detector.detect_all_tools.side_effect = _detect_all + detector._set_canonical_vscode_copilot.return_value = None + detector._set_canonical_augment_surface.return_value = None + detector.process_single_tool.side_effect = lambda t: t + detector.filter_tool_projects_by_user.side_effect = lambda t, _h: t + detector.generate_single_tool_report.side_effect = ( + lambda tool, device_id, home_user, system_user=None, run_id=None: {"tools": [tool]} + ) + + dc = Mock() + dc.acquire_lock.return_value = "acquired" + dc.heartbeat_start.return_value = Mock() + dc.get_cached_hash.return_value = None + dc.update_tool.return_value = None + dc.UNBOUND_DIR = "/tmp/unbound-test" + dc.last_lock_error = None + + captured = {} + + def _send_scan_event(domain, api_key, device_id, run_id, scan_event, app_name=None, **kw): + if scan_event == "completed": + captured["manifest"] = kw.get("manifest") + return (True, None) + + with patch.object(adm.platform, "system", return_value="Darwin"), \ + patch.object(adm, "AIToolsDetector", return_value=detector), \ + patch.object(adm, "discovery_cache", dc), \ + patch.object(adm, "get_all_users_macos", return_value=["alice", "bob"]), \ + patch.object(adm, "compute_payload_hash", side_effect=lambda t: "h-" + t["name"]), \ + patch.object(adm, "send_report_to_backend", return_value=(True, False)), \ + patch.object(adm, "send_scan_event", side_effect=_send_scan_event), \ + patch.object(adm, "send_discovery_metrics", Mock()), \ + patch.object(adm, "load_pending_reports", return_value=[]), \ + patch.object(adm, "save_failed_reports", Mock()), \ + patch.object(adm, "report_to_sentry", Mock()), \ + patch.object(utils_mod, "_SENTRY_DSN", ""), \ + patch.object(sys, "argv", self.argv): + try: + adm.main() + except SystemExit: + pass + + pairs = {(e["home_user"], e["tool_name"]) for e in captured["manifest"]} + self.assertEqual(pairs, {("alice", "ToolA"), ("bob", "ToolB")}) + self.assertNotIn(("bob", "ToolA"), pairs) + self.assertNotIn(("alice", "ToolB"), pairs) + + +class TestExtensionDetectorPerUserScoping(unittest.TestCase): + """Extension detectors must scope to the per-user home set by detect_tool_for_user, so an + elevated multi-user scan can't attribute one user's extension to another (phantom ownership).""" + + def test_roo_detect_scopes_to_set_user_home(self): + from scripts.coding_discovery_tools.macos.roo_code.roo_code import MacOSRooDetector + det = MacOSRooDetector() + det.user_home = Path("/Users/alice") + with patch.object(det, "_detect_roo_for_user", return_value=[{"name": "Roo Code (Cursor)"}]) as m: + result = det.detect() + # Called exactly once with the SET home (not Path.home(), not per-/Users enumeration). + m.assert_called_once_with(Path("/Users/alice")) + self.assertEqual(result, [{"name": "Roo Code (Cursor)"}]) + + +class TestJetBrainsNamingDeterminism(unittest.TestCase): + """The JetBrains tool name is the backend prune key (matched exactly vs the manifest), so it + must exclude version and license/plan — otherwise a version bump or Free<->Licensed change + would orphan the install row and wrongly prune it. Fails if a change re-embeds them in the name. + """ + + def setUp(self): + self.det = MacOSJetBrainsDetector() + + def test_display_name_is_version_free_and_stable_across_bumps(self): + for folder in ("PyCharm2025.3", "PyCharm2025.3.1", "PyCharm2026.1"): + name, version = self.det._parse_ide_name_and_version(folder) + self.assertEqual(name, "PyCharm", f"{folder} must map to stable 'PyCharm'") + self.assertNotIn(version, name, "version must not leak into the display name") + self.assertEqual( + self.det._parse_ide_name_and_version("IntelliJIdea2025.3")[0], "IntelliJ IDEA" + ) + + def test_mapping_values_carry_no_version_or_plan(self): + for _prefix, name in MacOSJetBrainsDetector.IDE_NAME_MAPPING.items(): + self.assertNotRegex(name, r"\d", f"{name!r} must not embed a version digit") + self.assertNotIn("(", name, f"{name!r} must not embed a (plan) suffix") + + def test_detected_tool_name_excludes_version_and_plan(self): + # detect() sets name = display_name ONLY; version and plan stay in separate fields. + fake_ide = { + "display_name": "PyCharm", "version": "2025.3.1", "plan": "Licensed", + "config_path": "/nonexistent/pycharm", "folder_name": "PyCharm2025.3.1", + } + with patch.object(self.det, "_scan_for_ides", return_value=[fake_ide]), \ + patch.object(self.det, "_get_plugins", return_value=[]): + tools = self.det.detect() + self.assertEqual(tools[0]["name"], "PyCharm", "prune key (name) must be the bare display_name") + self.assertNotIn("2025", tools[0]["name"]) + self.assertNotIn("Licensed", tools[0]["name"]) + self.assertEqual(tools[0]["version"], "2025.3.1") + self.assertEqual(tools[0]["plan"], "Licensed") + + +if __name__ == "__main__": + unittest.main()