From 7a072a6faa6d344dd592499161925f130ea96c7c Mon Sep 17 00:00:00 2001
From: Scott Beddall <scbedd@microsoft.com>
Date: Thu, 28 Aug 2025 01:03:43 +0000
Subject: [PATCH 01/12] saving progress before chatgpt starts

---
 eng/scripts/dispatch_checks.py                | 152 ++++++++++++++++++
 .../ci_tools/scenario/generation.py           |   2 +-
 2 files changed, 153 insertions(+), 1 deletion(-)
 create mode 100644 eng/scripts/dispatch_checks.py

diff --git a/eng/scripts/dispatch_checks.py b/eng/scripts/dispatch_checks.py
new file mode 100644
index 000000000000..26d2656059af
--- /dev/null
+++ b/eng/scripts/dispatch_checks.py
@@ -0,0 +1,152 @@
+import argparse
+import os
+import logging
+import sys
+
+from subprocess import check_call
+
+from ci_tools.functions import discover_targeted_packages
+from ci_tools.scenario import install_into_venv
+from ci_tools.functions import get_venv_call
+from ci_tools.variables import in_ci
+from ci_tools.scenario.generation import build_whl_for_req
+
+root_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+
+
+def create_venv(venv_path: str) -> None:
+    venv_creation_command = get_venv_call()
+    check_call(venv_creation_command + [venv_path])
+
+    # TODO: go to the prebuilt wheel dir if that exists and retrieve azure-sdk-tools from THAT if it exists
+    # this is ok for now though.
+    install_into_venv(venv_path, os.path.join(root_dir, "eng/tools/azure-sdk-tools"), False, "build")
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description="""
+This script is the single point for all checks invoked by CI within this repo. It works in two phases.
+    1. Identify which packages in the repo are in scope for this script invocation, based on a glob string and a service directory.
+    2. Invoke one or multiple `tox` environments for each package identified as in scope.
+
+In the case of an environment invoking `pytest`, results can be collected in a junit xml file, and test markers can be selected via --mark_arg.
+"""
+    )
+
+    parser.add_argument(
+        "glob_string",
+        nargs="?",
+        help=(
+            "A comma separated list of glob strings that will target the top level directories that contain packages."
+            'Examples: All = "azure-*", Single = "azure-keyvault", Targeted Multiple = "azure-keyvault,azure-mgmt-resource"'
+        ),
+    )
+
+    parser.add_argument(
+        "--junitxml",
+        dest="test_results",
+        help=(
+            "The output path for the test results file of invoked checks."
+            'Example: --junitxml="junit/test-results.xml"'
+        ),
+    )
+
+    parser.add_argument(
+        "--mark_arg",
+        dest="mark_arg",
+        help=(
+            'The complete argument for `pytest -m "<input>"`. This can be used to exclude or include specific pytest markers.'
+            '--mark_arg="not cosmosEmulator"'
+        ),
+    )
+
+    parser.add_argument("--disablecov", help=("Flag. Disables code coverage."), action="store_true")
+
+    parser.add_argument(
+        "--service",
+        help=("Name of service directory (under sdk/) to test. Example: --service applicationinsights"),
+    )
+
+    parser.add_argument(
+        "-c",
+        "--checks",
+        dest="checks_list",
+        help="Specific set of named environments to execute",
+    )
+
+    parser.add_argument(
+        "-w",
+        "--wheel_dir",
+        dest="wheel_dir",
+        help="Location for prebuilt artifacts (if any)",
+    )
+
+    parser.add_argument(
+        "-i",
+        "--injected-packages",
+        dest="injected_packages",
+        default="",
+        help="Comma or space-separated list of packages that should be installed prior to dev_requirements. If local path, should be absolute.",
+    )
+
+    parser.add_argument(
+        "--filter-type",
+        dest="filter_type",
+        default="Build",
+        help="Filter type to identify eligible packages. for e.g. packages filtered in Build can pass filter type as Build,",
+        choices=["Build", "Docs", "Regression", "Omit_management", "None"],
+    )
+
+    parser.add_argument(
+        "-d",
+        "--dest-dir",
+        dest="dest_dir",
+        help="Location to generate any output files(if any). For e.g. apiview stub file",
+    )
+
+    args = parser.parse_args()
+
+    # We need to support both CI builds of everything and individual service
+    # folders. This logic allows us to do both.
+    if args.service and args.service != "auto":
+        service_dir = os.path.join("sdk", args.service)
+        target_dir = os.path.join(root_dir, service_dir)
+    else:
+        target_dir = root_dir
+
+    logging.info(f"Beginning discovery for {args.service} and root dir {root_dir}. Resolving to {target_dir}.")
+
+    if args.filter_type == "None":
+        args.filter_type = "Build"
+        compatibility_filter = False
+    else:
+        compatibility_filter = True
+
+    targeted_packages = discover_targeted_packages(
+        args.glob_string, target_dir, "", args.filter_type, compatibility_filter
+    )
+
+    if len(targeted_packages) == 0:
+        logging.info(f"No packages collected for targeting string {args.glob_string} and root dir {root_dir}. Exit 0.")
+        exit(0)
+
+    print(f"Executing checks with the executable {sys.executable}.")
+    print(f"Packages targeted: {targeted_packages}")
+
+    if args.wheel_dir:
+        os.environ["PREBUILT_WHEEL_DIR"] = args.wheel_dir
+    else:
+        os.environ["PREBUILT_WHEEL_DIR"] = os.path.join(root_dir, ".wheels")
+
+    if in_ci():
+        # prepare a build of eng/tools/azure-sdk-tools
+        build_whl_for_req("eng/tools/azure-sdk-tools", root_dir, os.environ.get("PREBUILT_WHEEL_DIR"))
+
+    # so if we have checks whl,import_all and selected package paths `sdk/core/azure-core`, `sdk/storage/azure-storage-blob` we should
+    # shell out to `azypysdk <checkname>` with cwd of the package directory, which is what is in `targeted_packages` array
+    # each individual thread may need to re-invoke if they need to self-isolate themselves, but we don't have to worry about that.
+
+    for package in targeted_packages:
+        for check_name in args.checks_list.split(","):
+            pass
+
diff --git a/eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py b/eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py
index a216f927992a..cfc13c9e246f 100644
--- a/eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py
+++ b/eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py
@@ -316,7 +316,7 @@ def build_whl_for_req(req: str, package_path: str, wheel_dir: Optional[str]) ->
     """Builds a whl from the dev_requirements file.
 
     :param str req: a requirement from the dev_requirements.txt
-    :param str package_path: the absolute path to the package's root
+    :param str package_path: the absolute path to the package's root (used as relative path root)
     :param Optional[str] wheel_dir: the absolute path to the prebuilt wheel directory
     :return: The absolute path to the whl built or the requirement if a third-party package
     """

From 36938ba20ea660ceed8d3b934a4722a27c338631 Mon Sep 17 00:00:00 2001
From: Scott Beddall <scbedd@microsoft.com>
Date: Thu, 28 Aug 2025 01:05:58 +0000
Subject: [PATCH 02/12] saving progress on eng! time to debug this thing and
 see what falls apart

---
 eng/scripts/dispatch_checks.py | 150 ++++++++++++++++++++++++++++++++-
 1 file changed, 146 insertions(+), 4 deletions(-)

diff --git a/eng/scripts/dispatch_checks.py b/eng/scripts/dispatch_checks.py
index 26d2656059af..3b986e8aee25 100644
--- a/eng/scripts/dispatch_checks.py
+++ b/eng/scripts/dispatch_checks.py
@@ -1,9 +1,13 @@
 import argparse
+import asyncio
 import os
 import logging
 import sys
-
+import time
+import signal
+from dataclasses import dataclass
 from subprocess import check_call
+from typing import List
 
 from ci_tools.functions import discover_targeted_packages
 from ci_tools.scenario import install_into_venv
@@ -22,6 +26,123 @@ def create_venv(venv_path: str) -> None:
     # this is ok for now though.
     install_into_venv(venv_path, os.path.join(root_dir, "eng/tools/azure-sdk-tools"), False, "build")
 
+@dataclass
+class CheckResult:
+    package: str
+    check: str
+    exit_code: int
+    duration: float
+    stdout: str
+    stderr: str
+
+
+async def run_check(semaphore: asyncio.Semaphore, package: str, check: str, base_args: List[str], idx: int, total: int) -> CheckResult:
+    """Run a single check (subprocess) within a concurrency semaphore, capturing output and timing.
+
+    Args:
+        semaphore: Concurrency limiter.
+        package: Absolute path to package directory.
+        check: The check (subcommand) name for azpysdk CLI.
+        base_args: Common argument list prefix (python -m azpysdk.main ...).
+        idx: Sequence number for logging.
+        total: Total number of tasks.
+    """
+    async with semaphore:
+        start = time.time()
+        cmd = base_args + [check, package]
+        logging.info(f"[START {idx}/{total}] {check} :: {package}\nCMD: {' '.join(cmd)}")
+        try:
+            proc = await asyncio.create_subprocess_exec(
+                *cmd,
+                cwd=package,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+        except Exception as ex:  # subprocess failed to launch
+            logging.error(f"Failed to start check {check} for {package}: {ex}")
+            return CheckResult(package, check, 127, 0.0, "", str(ex))
+
+        stdout_b, stderr_b = await proc.communicate()
+        duration = time.time() - start
+        stdout = stdout_b.decode(errors="replace")
+        stderr = stderr_b.decode(errors="replace")
+        exit_code = proc.returncode or 0
+        status = "OK" if exit_code == 0 else f"FAIL({exit_code})"
+        logging.info(f"[END   {idx}/{total}] {check} :: {package} -> {status} in {duration:.2f}s")
+        # Print captured output after completion to avoid interleaving
+        header = f"===== OUTPUT: {check} :: {package} (exit {exit_code}) ====="
+        trailer = "=" * len(header)
+        if stdout:
+            print(header)
+            print(stdout.rstrip())
+            print(trailer)
+        if stderr:
+            print(header.replace('OUTPUT', 'STDERR'))
+            print(stderr.rstrip())
+            print(trailer)
+        return CheckResult(package, check, exit_code, duration, stdout, stderr)
+
+
+def summarize(results: List[CheckResult]) -> int:
+    # Compute column widths
+    pkg_w = max((len(r.package) for r in results), default=7)
+    chk_w = max((len(r.check) for r in results), default=5)
+    header = f"{'PACKAGE'.ljust(pkg_w)}  {'CHECK'.ljust(chk_w)}  STATUS  DURATION(s)"
+    print("\n=== SUMMARY ===")
+    print(header)
+    print("-" * len(header))
+    for r in sorted(results, key=lambda x: (x.exit_code != 0, x.package, x.check)):
+        status = "OK" if r.exit_code == 0 else f"FAIL({r.exit_code})"
+        print(f"{r.package.ljust(pkg_w)}  {r.check.ljust(chk_w)}  {status.ljust(8)}  {r.duration:>10.2f}")
+    worst = max((r.exit_code for r in results), default=0)
+    failed = [r for r in results if r.exit_code != 0]
+    print(f"\nTotal checks: {len(results)} | Failed: {len(failed)} | Worst exit code: {worst}")
+    return worst
+
+
+async def run_all_checks(packages, checks, max_parallel):
+    base_args = [sys.executable, "-m", "azpysdk.main"]
+    tasks = []
+    semaphore = asyncio.Semaphore(max_parallel)
+    combos = [(p, c) for p in packages for c in checks]
+    total = len(combos)
+    for idx, (package, check) in enumerate(combos, start=1):
+        tasks.append(asyncio.create_task(run_check(semaphore, package, check, base_args, idx, total)))
+
+    # Handle Ctrl+C gracefully
+    pending = set(tasks)
+    try:
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+    except KeyboardInterrupt:
+        logging.warning("KeyboardInterrupt received. Cancelling running checks...")
+        for t in pending:
+            t.cancel()
+        raise
+    # Normalize exceptions
+    norm_results: List[CheckResult] = []
+    for res, (package, check) in zip(results, combos):
+        if isinstance(res, CheckResult):
+            norm_results.append(res)
+        elif isinstance(res, Exception):
+            norm_results.append(CheckResult(package, check, 99, 0.0, "", str(res)))
+        else:
+            norm_results.append(CheckResult(package, check, 98, 0.0, "", f"Unknown result type: {res}"))
+    return summarize(norm_results)
+
+
+def configure_interrupt_handling():
+    # Ensure that a SIGINT propagates to asyncio tasks & subprocesses
+    def handler(signum, frame):  # noqa: D401
+        logging.warning(f"Received signal {signum}. Attempting graceful shutdown...")
+        # Let asyncio loop raise KeyboardInterrupt
+        raise KeyboardInterrupt
+
+    try:
+        signal.signal(signal.SIGINT, handler)
+    except (ValueError, AttributeError):  # not supported on some platforms/threads
+        pass
+
+
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(
         description="""
@@ -104,6 +225,14 @@ def create_venv(venv_path: str) -> None:
         help="Location to generate any output files(if any). For e.g. apiview stub file",
     )
 
+    parser.add_argument(
+        "--max-parallel",
+        dest="max_parallel",
+        type=int,
+        default=os.cpu_count() or 4,
+        help="Maximum number of concurrent checks (default: number of CPU cores).",
+    )
+
     args = parser.parse_args()
 
     # We need to support both CI builds of everything and individual service
@@ -146,7 +275,20 @@ def create_venv(venv_path: str) -> None:
     # shell out to `azypysdk <checkname>` with cwd of the package directory, which is what is in `targeted_packages` array
     # each individual thread may need to re-invoke if they need to self-isolate themselves, but we don't have to worry about that.
 
-    for package in targeted_packages:
-        for check_name in args.checks_list.split(","):
-            pass
+    # Prepare check list
+    raw_checks = (args.checks_list or "").split(",")
+    checks = [c.strip() for c in raw_checks if c and c.strip()]
+    if not checks:
+        logging.error("No valid checks provided via -c/--checks.")
+        sys.exit(2)
+
+    logging.info(f"Running {len(checks)} checks across {len(targeted_packages)} packages (max_parallel={args.max_parallel}).")
+
+    configure_interrupt_handling()
+    try:
+        exit_code = asyncio.run(run_all_checks(targeted_packages, checks, args.max_parallel))
+    except KeyboardInterrupt:
+        logging.error("Aborted by user.")
+        exit_code = 130
+    sys.exit(exit_code)
 

From bd8af223a5d250eff30aa76ccd456507fe8fd9aa Mon Sep 17 00:00:00 2001
From: Scott Beddall <scbedd@microsoft.com>
Date: Thu, 28 Aug 2025 18:02:52 +0000
Subject: [PATCH 03/12] handle venv abstraction

---
 eng/tools/azure-sdk-tools/azpysdk/Check.py | 46 +++++++++++++++++++++-
 eng/tools/azure-sdk-tools/azpysdk/main.py  | 43 +-------------------
 eng/tools/azure-sdk-tools/azpysdk/whl.py   | 25 +++++++++---
 3 files changed, 65 insertions(+), 49 deletions(-)

diff --git a/eng/tools/azure-sdk-tools/azpysdk/Check.py b/eng/tools/azure-sdk-tools/azpysdk/Check.py
index cd18731eed5b..0382b0263cc2 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/Check.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/Check.py
@@ -2,9 +2,20 @@
 import os
 import argparse
 import traceback
+import sys
+
 from typing import Sequence, Optional, List, Any
+from subprocess import check_call
+
 from ci_tools.parsing import ParsedSetup
-from ci_tools.functions import discover_targeted_packages
+from ci_tools.functions import discover_targeted_packages, get_venv_call
+from ci_tools.variables import discover_repo_root
+from ci_tools.scenario import install_into_venv, get_venv_python
+
+# right now, we are assuming you HAVE to be in the azure-sdk-tools repo
+# we assume this because we don't know how a dev has installed this package, and might be
+# being called from within a site-packages folder. Due to that, we can't trust the location of __file__
+REPO_ROOT = discover_repo_root()
 
 class Check(abc.ABC):
     """
@@ -34,13 +45,44 @@ def run(self, args: argparse.Namespace) -> int:
         """
         return 0
 
+    def handle_venv(self, isolate: bool, args: argparse.Namespace, venv_location: Optional[str]) -> str:
+        """Handle virtual environment commands."""
+        # if we have to isolate, return the new python exe that the rest of the checks should use
+        if (isolate):
+            # os.environ["AZURE_SDK_TOOLS_VENV"] = "1"
+
+            venv_cmd = get_venv_call()
+            if not venv_location:
+                venv_location = os.path.join(REPO_ROOT, f".venv_{args.command}")
+
+            # todo, make this a consistent directory based on the command
+            # I'm seriously thinking we should move handle_venv within each check's main(),
+            # which will mean that we will _know_ what folder we're in.
+            # however, that comes at the cost of not having every check be able to handle one or multiple packages
+            # I don't want to get into an isolation loop where every time we need a new venv, we create it, call it,
+            # and now as we foreach across the targeted packages we've lost our spot.
+            check_call(venv_cmd + [venv_location])
+
+            # TODO: we should reuse part of build_whl_for_req to integrate with PREBUILT_WHL_DIR so that we don't have to fresh build for each
+            # venv
+            install_into_venv(venv_location, os.path.join(REPO_ROOT, "eng/tools/azure-sdk-tools"), False, "build")
+            venv_python_exe = get_venv_python(venv_location)
+
+            # here is a newly prepped virtual environment (which includes azure-sdk-tools)
+            return venv_python_exe
+            # command_args = [venv_python_exe, "-m", "azpysdk.main"] + sys.argv[1:]
+            # check_call(command_args)
+
+        # if we don't need to isolate, just return the python executable that we're invoking
+        return sys.executable
+
     def get_targeted_directories(self, args: argparse.Namespace) -> List[ParsedSetup]:
         """
         Get the directories that are targeted for the check.
         """
         targeted: List[ParsedSetup] = []
         targeted_dir = os.getcwd()
- 
+
         if args.target == ".":
             try:
                 targeted.append(ParsedSetup.from_path(targeted_dir))
diff --git a/eng/tools/azure-sdk-tools/azpysdk/main.py b/eng/tools/azure-sdk-tools/azpysdk/main.py
index 59f830c205ee..ec555aa0d3a8 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/main.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/main.py
@@ -11,21 +11,11 @@
 import sys
 import os
 from typing import Sequence, Optional
-from subprocess import check_call
 
 from .whl import whl
 from .import_all import import_all
 from .mypy import mypy
 
-from ci_tools.scenario import install_into_venv, get_venv_python
-from ci_tools.functions import get_venv_call
-from ci_tools.variables import discover_repo_root
-
-# right now, we are assuming you HAVE to be in the azure-sdk-tools repo
-# we assume this because we don't know how a dev has installed this package, and might be
-# being called from within a site-packages folder. Due to that, we can't trust the location of __file__
-REPO_ROOT = discover_repo_root()
-
 __all__ = ["main", "build_parser"]
 __version__ = "0.0.0"
 
@@ -63,35 +53,7 @@ def build_parser() -> argparse.ArgumentParser:
 
     return parser
 
-def handle_venv(isolate: bool, args: argparse.Namespace) -> None:
-    """Handle virtual environment commands."""
-    # we are already in an isolated venv and so do not need to recurse
-    if(os.getenv("AZURE_SDK_TOOLS_VENV", None)):
-        return
-
-    # however, if we are not already in an isolated venv, and should be, then we need to
-    # call
-    if (isolate):
-        os.environ["AZURE_SDK_TOOLS_VENV"] = "1"
-
-        venv_cmd = get_venv_call()
-        venv_location = os.path.join(REPO_ROOT, f".venv_{args.command}")
-        # todo, make this a consistent directory based on the command
-        # I'm seriously thinking we should move handle_venv within each check's main(),
-        # which will mean that we will _know_ what folder we're in.
-        # however, that comes at the cost of not having every check be able to handle one or multiple packages
-        # I don't want to get into an isolation loop where every time we need a new venv, we create it, call it,
-        # and now as we foreach across the targeted packages we've lost our spot.
-        check_call(venv_cmd + [venv_location])
-
-        # now use the current virtual environment to install os.path.join(REPO_ROOT, eng/tools/azure-sdk-tools[build])
-        # into the NEW virtual env
-        install_into_venv(venv_location, os.path.join(REPO_ROOT, "eng/tools/azure-sdk-tools"), False, "build")
-        venv_python_exe = get_venv_python(venv_location)
-        command_args = [venv_python_exe, "-m", "azpysdk.main"] + sys.argv[1:]
-        check_call(command_args)
-
-def main(argv: Optional[Sequence[str]] = None) -> int:#
+def main(argv: Optional[Sequence[str]] = None) -> int:
     """CLI entrypoint.
 
     Args:
@@ -108,9 +70,6 @@ def main(argv: Optional[Sequence[str]] = None) -> int:#
         return 1
 
     try:
-        # scbedd 8/25 I'm betting that this would be best placed within the check itself,
-        # but leaving this for now
-        handle_venv(args.isolate, args)
         result = args.func(args)
         return int(result or 0)
     except KeyboardInterrupt:
diff --git a/eng/tools/azure-sdk-tools/azpysdk/whl.py b/eng/tools/azure-sdk-tools/azpysdk/whl.py
index b6d9bc4f8f93..1f524b4ea2b5 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/whl.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/whl.py
@@ -3,8 +3,8 @@
 import tempfile
 import os
 from typing import Optional, List, Any
-from pytest import main as pytest_main
 import sys
+from subprocess import check_call
 
 from .Check import Check
 
@@ -39,11 +39,15 @@ def run(self, args: argparse.Namespace) -> int:
 
         for parsed in targeted:
             pkg = parsed.folder
+            venv_location = os.path.join(parsed.folder, f".venv_{args.command}")
+            # if isolation is required, the executable we get back will align with the venv
+            # otherwise we'll just get sys.executable and install in current
+            executable = self.handle_venv(args.isolate, args, venv_location=venv_location)
 
             dev_requirements = os.path.join(pkg, "dev_requirements.txt")
 
             if os.path.exists(dev_requirements):
-                pip_install([f"-r", f"{dev_requirements}"], True, sys.executable)
+                pip_install([f"-r", f"{dev_requirements}"], True, executable)
 
             staging_area = tempfile.mkdtemp()
             create_package_and_install(
@@ -55,13 +59,24 @@ def run(self, args: argparse.Namespace) -> int:
                 force_create=False,
                 package_type="wheel",
                 pre_download_disabled=False,
+                python_executable=executable
             )
 
             # todo, come up with a good pattern for passing all the additional args after -- to pytest
             logging.info(f"Invoke pytest for {pkg}")
-
-            exit_code = pytest_main(
-                [pkg]
+            exit_code = check_call(
+                [executable, "-m", "pytest", pkg] + (["-m", args.mark_arg] if args.mark_arg else []) + [
+                    "-rsfE",
+                    f"--junitxml={pkg}/test-junit-{args.command}.xml",
+                    "--verbose",
+                    "--cov-branch",
+                    "--durations=10",
+                    "--ignore=azure",
+                    "--ignore-glob=.venv*",
+                    "--ignore=build",
+                    "--ignore=.eggs",
+                    "--ignore=samples"
+                ]
             )
 
             if exit_code != 0:

From fc2f8fdd5638e81163185482079ea3878a0f66f8 Mon Sep 17 00:00:00 2001
From: Scott Beddall <scbedd@microsoft.com>
Date: Thu, 28 Aug 2025 18:19:30 +0000
Subject: [PATCH 04/12] much closer, just need to figure out why this isn't
 calling properly

---
 eng/tools/azure-sdk-tools/azpysdk/whl.py        | 11 +++++++++--
 eng/tools/azure-sdk-tools/ci_tools/functions.py |  7 +++++--
 2 files changed, 14 insertions(+), 4 deletions(-)

diff --git a/eng/tools/azure-sdk-tools/azpysdk/whl.py b/eng/tools/azure-sdk-tools/azpysdk/whl.py
index 1f524b4ea2b5..8b615fb8dbb0 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/whl.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/whl.py
@@ -44,11 +44,16 @@ def run(self, args: argparse.Namespace) -> int:
             # otherwise we'll just get sys.executable and install in current
             executable = self.handle_venv(args.isolate, args, venv_location=venv_location)
 
+            print(f"Invoking check with {executable}")
             dev_requirements = os.path.join(pkg, "dev_requirements.txt")
 
             if os.path.exists(dev_requirements):
-                pip_install([f"-r", f"{dev_requirements}"], True, executable)
+                print(f"Installing dev_requirements at {dev_requirements}")
+                pip_install([f"-r", f"{dev_requirements}"], True, executable, pkg)
+            else:
+                print("Skipping installing dev_requirements")
 
+            # TODO: make the staging area a folder under the venv_location
             staging_area = tempfile.mkdtemp()
             create_package_and_install(
                 distribution_directory=staging_area,
@@ -64,8 +69,9 @@ def run(self, args: argparse.Namespace) -> int:
 
             # todo, come up with a good pattern for passing all the additional args after -- to pytest
             logging.info(f"Invoke pytest for {pkg}")
+            # + (["-m", args.mark_arg] if args.mark_arg else []) +
             exit_code = check_call(
-                [executable, "-m", "pytest", pkg] + (["-m", args.mark_arg] if args.mark_arg else []) + [
+                [executable, "-m", "pytest", "."] + [
                     "-rsfE",
                     f"--junitxml={pkg}/test-junit-{args.command}.xml",
                     "--verbose",
@@ -77,6 +83,7 @@ def run(self, args: argparse.Namespace) -> int:
                     "--ignore=.eggs",
                     "--ignore=samples"
                 ]
+                , cwd=pkg
             )
 
             if exit_code != 0:
diff --git a/eng/tools/azure-sdk-tools/ci_tools/functions.py b/eng/tools/azure-sdk-tools/ci_tools/functions.py
index 999bb919d3c5..4642d902b630 100644
--- a/eng/tools/azure-sdk-tools/ci_tools/functions.py
+++ b/eng/tools/azure-sdk-tools/ci_tools/functions.py
@@ -438,7 +438,7 @@ def find_sdist(dist_dir: str, pkg_name: str, pkg_version: str) -> Optional[str]:
 
 
 def pip_install(
-    requirements: List[str], include_dependencies: bool = True, python_executable: Optional[str] = None
+    requirements: List[str], include_dependencies: bool = True, python_executable: Optional[str] = None, cwd: Optional[str] = None
 ) -> bool:
     """
     Attempts to invoke an install operation using the invoking python's pip. Empty requirements are auto-success.
@@ -454,7 +454,10 @@ def pip_install(
         return True
 
     try:
-        subprocess.check_call(command)
+        if cwd:
+            subprocess.check_call(command, cwd=cwd)
+        else:
+            subprocess.check_call(command)
     except subprocess.CalledProcessError as f:
         return False
 

From 9daa786b6ec3ad5bbe9160599a23acaa0e611acf Mon Sep 17 00:00:00 2001
From: Scott Beddall <scbedd@microsoft.com>
Date: Thu, 28 Aug 2025 18:49:13 +0000
Subject: [PATCH 05/12] uv pip list instead of python -m pip list

---
 eng/tools/azure-sdk-tools/ci_tools/functions.py           | 4 +++-
 eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py | 1 +
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/eng/tools/azure-sdk-tools/ci_tools/functions.py b/eng/tools/azure-sdk-tools/ci_tools/functions.py
index 4642d902b630..967d09d105d0 100644
--- a/eng/tools/azure-sdk-tools/ci_tools/functions.py
+++ b/eng/tools/azure-sdk-tools/ci_tools/functions.py
@@ -491,8 +491,10 @@ def get_pip_list_output(python_executable: Optional[str] = None):
     """Uses the invoking python executable to get the output from pip list."""
     exe = python_executable or sys.executable
 
+    pip_cmd = get_pip_command(exe)
+
     out = subprocess.Popen(
-        [exe, "-m", "pip", "list", "--disable-pip-version-check", "--format", "freeze"],
+        pip_cmd + ["list", "--disable-pip-version-check", "--format", "freeze"],
         stdout=subprocess.PIPE,
         stderr=subprocess.STDOUT,
     )
diff --git a/eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py b/eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py
index cfc13c9e246f..ba592d29f8d8 100644
--- a/eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py
+++ b/eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py
@@ -130,6 +130,7 @@ def create_package_and_install(
                         addition_necessary = True
                         # get all installed packages
                         installed_pkgs = get_pip_list_output(python_exe)
+                        breakpoint()
                         logging.info("Installed packages: {}".format(installed_pkgs))
 
                         # parse the specifier

From faa6feff8fab79a45bcf5b66da69cd3fe466b4e1 Mon Sep 17 00:00:00 2001
From: Scott Beddall <scbedd@microsoft.com>
Date: Thu, 28 Aug 2025 19:11:01 +0000
Subject: [PATCH 06/12] call in properly

---
 eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py b/eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py
index ba592d29f8d8..cfc13c9e246f 100644
--- a/eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py
+++ b/eng/tools/azure-sdk-tools/ci_tools/scenario/generation.py
@@ -130,7 +130,6 @@ def create_package_and_install(
                         addition_necessary = True
                         # get all installed packages
                         installed_pkgs = get_pip_list_output(python_exe)
-                        breakpoint()
                         logging.info("Installed packages: {}".format(installed_pkgs))
 
                         # parse the specifier

From 1aa748e470015eb4bf0b25e3ce81ab3e45429065 Mon Sep 17 00:00:00 2001
From: Scott Beddall <scbedd@microsoft.com>
Date: Thu, 28 Aug 2025 20:07:06 +0000
Subject: [PATCH 07/12] save progress

---
 eng/tools/azure-sdk-tools/azpysdk/whl.py | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/eng/tools/azure-sdk-tools/azpysdk/whl.py b/eng/tools/azure-sdk-tools/azpysdk/whl.py
index 8b615fb8dbb0..d5d2d127772a 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/whl.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/whl.py
@@ -25,7 +25,7 @@ def register(self, subparsers: "argparse._SubParsersAction", parent_parsers: Opt
         parents = parent_parsers or []
         p = subparsers.add_parser("whl", parents=parents, help="Run the whl check")
         p.set_defaults(func=self.run)
-        # Add any additional arguments specific to WhlCheck here (do not re-add common handled by parents)
+        # TODO add mark_args, and other parameters
 
     def run(self, args: argparse.Namespace) -> int:
         """Run the whl check command."""
@@ -43,6 +43,8 @@ def run(self, args: argparse.Namespace) -> int:
             # if isolation is required, the executable we get back will align with the venv
             # otherwise we'll just get sys.executable and install in current
             executable = self.handle_venv(args.isolate, args, venv_location=venv_location)
+            staging_directory = os.path.join(venv_location, ".staging")
+            os.makedirs(staging_directory, exist_ok=True)
 
             print(f"Invoking check with {executable}")
             dev_requirements = os.path.join(pkg, "dev_requirements.txt")
@@ -53,23 +55,21 @@ def run(self, args: argparse.Namespace) -> int:
             else:
                 print("Skipping installing dev_requirements")
 
-            # TODO: make the staging area a folder under the venv_location
-            staging_area = tempfile.mkdtemp()
             create_package_and_install(
-                distribution_directory=staging_area,
+                distribution_directory=staging_directory,
                 target_setup=pkg,
                 skip_install=False,
                 cache_dir=None,
-                work_dir=staging_area,
+                work_dir=staging_directory,
                 force_create=False,
                 package_type="wheel",
                 pre_download_disabled=False,
                 python_executable=executable
             )
 
-            # todo, come up with a good pattern for passing all the additional args after -- to pytest
+            # TODO: split sys.argv[1:] on -- and pass in everything after the -- as additional arguments
+            # TODO: handle mark_args
             logging.info(f"Invoke pytest for {pkg}")
-            # + (["-m", args.mark_arg] if args.mark_arg else []) +
             exit_code = check_call(
                 [executable, "-m", "pytest", "."] + [
                     "-rsfE",

From c9aea81d6f28c373d60351cfbe761f63c42e8edf Mon Sep 17 00:00:00 2001
From: Scott Beddall <scbedd@microsoft.com>
Date: Thu, 28 Aug 2025 20:54:24 +0000
Subject: [PATCH 08/12] changes to processing of venv

---
 eng/scripts/dispatch_checks.py                | 294 ------------------
 eng/tools/azure-sdk-tools/azpysdk/Check.py    |  20 +-
 .../azure-sdk-tools/azpysdk/import_all.py     |  10 +-
 eng/tools/azure-sdk-tools/azpysdk/mypy.py     |  27 +-
 eng/tools/azure-sdk-tools/azpysdk/whl.py      |   9 +-
 5 files changed, 35 insertions(+), 325 deletions(-)
 delete mode 100644 eng/scripts/dispatch_checks.py

diff --git a/eng/scripts/dispatch_checks.py b/eng/scripts/dispatch_checks.py
deleted file mode 100644
index 3b986e8aee25..000000000000
--- a/eng/scripts/dispatch_checks.py
+++ /dev/null
@@ -1,294 +0,0 @@
-import argparse
-import asyncio
-import os
-import logging
-import sys
-import time
-import signal
-from dataclasses import dataclass
-from subprocess import check_call
-from typing import List
-
-from ci_tools.functions import discover_targeted_packages
-from ci_tools.scenario import install_into_venv
-from ci_tools.functions import get_venv_call
-from ci_tools.variables import in_ci
-from ci_tools.scenario.generation import build_whl_for_req
-
-root_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
-
-
-def create_venv(venv_path: str) -> None:
-    venv_creation_command = get_venv_call()
-    check_call(venv_creation_command + [venv_path])
-
-    # TODO: go to the prebuilt wheel dir if that exists and retrieve azure-sdk-tools from THAT if it exists
-    # this is ok for now though.
-    install_into_venv(venv_path, os.path.join(root_dir, "eng/tools/azure-sdk-tools"), False, "build")
-
-@dataclass
-class CheckResult:
-    package: str
-    check: str
-    exit_code: int
-    duration: float
-    stdout: str
-    stderr: str
-
-
-async def run_check(semaphore: asyncio.Semaphore, package: str, check: str, base_args: List[str], idx: int, total: int) -> CheckResult:
-    """Run a single check (subprocess) within a concurrency semaphore, capturing output and timing.
-
-    Args:
-        semaphore: Concurrency limiter.
-        package: Absolute path to package directory.
-        check: The check (subcommand) name for azpysdk CLI.
-        base_args: Common argument list prefix (python -m azpysdk.main ...).
-        idx: Sequence number for logging.
-        total: Total number of tasks.
-    """
-    async with semaphore:
-        start = time.time()
-        cmd = base_args + [check, package]
-        logging.info(f"[START {idx}/{total}] {check} :: {package}\nCMD: {' '.join(cmd)}")
-        try:
-            proc = await asyncio.create_subprocess_exec(
-                *cmd,
-                cwd=package,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
-            )
-        except Exception as ex:  # subprocess failed to launch
-            logging.error(f"Failed to start check {check} for {package}: {ex}")
-            return CheckResult(package, check, 127, 0.0, "", str(ex))
-
-        stdout_b, stderr_b = await proc.communicate()
-        duration = time.time() - start
-        stdout = stdout_b.decode(errors="replace")
-        stderr = stderr_b.decode(errors="replace")
-        exit_code = proc.returncode or 0
-        status = "OK" if exit_code == 0 else f"FAIL({exit_code})"
-        logging.info(f"[END   {idx}/{total}] {check} :: {package} -> {status} in {duration:.2f}s")
-        # Print captured output after completion to avoid interleaving
-        header = f"===== OUTPUT: {check} :: {package} (exit {exit_code}) ====="
-        trailer = "=" * len(header)
-        if stdout:
-            print(header)
-            print(stdout.rstrip())
-            print(trailer)
-        if stderr:
-            print(header.replace('OUTPUT', 'STDERR'))
-            print(stderr.rstrip())
-            print(trailer)
-        return CheckResult(package, check, exit_code, duration, stdout, stderr)
-
-
-def summarize(results: List[CheckResult]) -> int:
-    # Compute column widths
-    pkg_w = max((len(r.package) for r in results), default=7)
-    chk_w = max((len(r.check) for r in results), default=5)
-    header = f"{'PACKAGE'.ljust(pkg_w)}  {'CHECK'.ljust(chk_w)}  STATUS  DURATION(s)"
-    print("\n=== SUMMARY ===")
-    print(header)
-    print("-" * len(header))
-    for r in sorted(results, key=lambda x: (x.exit_code != 0, x.package, x.check)):
-        status = "OK" if r.exit_code == 0 else f"FAIL({r.exit_code})"
-        print(f"{r.package.ljust(pkg_w)}  {r.check.ljust(chk_w)}  {status.ljust(8)}  {r.duration:>10.2f}")
-    worst = max((r.exit_code for r in results), default=0)
-    failed = [r for r in results if r.exit_code != 0]
-    print(f"\nTotal checks: {len(results)} | Failed: {len(failed)} | Worst exit code: {worst}")
-    return worst
-
-
-async def run_all_checks(packages, checks, max_parallel):
-    base_args = [sys.executable, "-m", "azpysdk.main"]
-    tasks = []
-    semaphore = asyncio.Semaphore(max_parallel)
-    combos = [(p, c) for p in packages for c in checks]
-    total = len(combos)
-    for idx, (package, check) in enumerate(combos, start=1):
-        tasks.append(asyncio.create_task(run_check(semaphore, package, check, base_args, idx, total)))
-
-    # Handle Ctrl+C gracefully
-    pending = set(tasks)
-    try:
-        results = await asyncio.gather(*tasks, return_exceptions=True)
-    except KeyboardInterrupt:
-        logging.warning("KeyboardInterrupt received. Cancelling running checks...")
-        for t in pending:
-            t.cancel()
-        raise
-    # Normalize exceptions
-    norm_results: List[CheckResult] = []
-    for res, (package, check) in zip(results, combos):
-        if isinstance(res, CheckResult):
-            norm_results.append(res)
-        elif isinstance(res, Exception):
-            norm_results.append(CheckResult(package, check, 99, 0.0, "", str(res)))
-        else:
-            norm_results.append(CheckResult(package, check, 98, 0.0, "", f"Unknown result type: {res}"))
-    return summarize(norm_results)
-
-
-def configure_interrupt_handling():
-    # Ensure that a SIGINT propagates to asyncio tasks & subprocesses
-    def handler(signum, frame):  # noqa: D401
-        logging.warning(f"Received signal {signum}. Attempting graceful shutdown...")
-        # Let asyncio loop raise KeyboardInterrupt
-        raise KeyboardInterrupt
-
-    try:
-        signal.signal(signal.SIGINT, handler)
-    except (ValueError, AttributeError):  # not supported on some platforms/threads
-        pass
-
-
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser(
-        description="""
-This script is the single point for all checks invoked by CI within this repo. It works in two phases.
-    1. Identify which packages in the repo are in scope for this script invocation, based on a glob string and a service directory.
-    2. Invoke one or multiple `tox` environments for each package identified as in scope.
-
-In the case of an environment invoking `pytest`, results can be collected in a junit xml file, and test markers can be selected via --mark_arg.
-"""
-    )
-
-    parser.add_argument(
-        "glob_string",
-        nargs="?",
-        help=(
-            "A comma separated list of glob strings that will target the top level directories that contain packages."
-            'Examples: All = "azure-*", Single = "azure-keyvault", Targeted Multiple = "azure-keyvault,azure-mgmt-resource"'
-        ),
-    )
-
-    parser.add_argument(
-        "--junitxml",
-        dest="test_results",
-        help=(
-            "The output path for the test results file of invoked checks."
-            'Example: --junitxml="junit/test-results.xml"'
-        ),
-    )
-
-    parser.add_argument(
-        "--mark_arg",
-        dest="mark_arg",
-        help=(
-            'The complete argument for `pytest -m "<input>"`. This can be used to exclude or include specific pytest markers.'
-            '--mark_arg="not cosmosEmulator"'
-        ),
-    )
-
-    parser.add_argument("--disablecov", help=("Flag. Disables code coverage."), action="store_true")
-
-    parser.add_argument(
-        "--service",
-        help=("Name of service directory (under sdk/) to test. Example: --service applicationinsights"),
-    )
-
-    parser.add_argument(
-        "-c",
-        "--checks",
-        dest="checks_list",
-        help="Specific set of named environments to execute",
-    )
-
-    parser.add_argument(
-        "-w",
-        "--wheel_dir",
-        dest="wheel_dir",
-        help="Location for prebuilt artifacts (if any)",
-    )
-
-    parser.add_argument(
-        "-i",
-        "--injected-packages",
-        dest="injected_packages",
-        default="",
-        help="Comma or space-separated list of packages that should be installed prior to dev_requirements. If local path, should be absolute.",
-    )
-
-    parser.add_argument(
-        "--filter-type",
-        dest="filter_type",
-        default="Build",
-        help="Filter type to identify eligible packages. for e.g. packages filtered in Build can pass filter type as Build,",
-        choices=["Build", "Docs", "Regression", "Omit_management", "None"],
-    )
-
-    parser.add_argument(
-        "-d",
-        "--dest-dir",
-        dest="dest_dir",
-        help="Location to generate any output files(if any). For e.g. apiview stub file",
-    )
-
-    parser.add_argument(
-        "--max-parallel",
-        dest="max_parallel",
-        type=int,
-        default=os.cpu_count() or 4,
-        help="Maximum number of concurrent checks (default: number of CPU cores).",
-    )
-
-    args = parser.parse_args()
-
-    # We need to support both CI builds of everything and individual service
-    # folders. This logic allows us to do both.
-    if args.service and args.service != "auto":
-        service_dir = os.path.join("sdk", args.service)
-        target_dir = os.path.join(root_dir, service_dir)
-    else:
-        target_dir = root_dir
-
-    logging.info(f"Beginning discovery for {args.service} and root dir {root_dir}. Resolving to {target_dir}.")
-
-    if args.filter_type == "None":
-        args.filter_type = "Build"
-        compatibility_filter = False
-    else:
-        compatibility_filter = True
-
-    targeted_packages = discover_targeted_packages(
-        args.glob_string, target_dir, "", args.filter_type, compatibility_filter
-    )
-
-    if len(targeted_packages) == 0:
-        logging.info(f"No packages collected for targeting string {args.glob_string} and root dir {root_dir}. Exit 0.")
-        exit(0)
-
-    print(f"Executing checks with the executable {sys.executable}.")
-    print(f"Packages targeted: {targeted_packages}")
-
-    if args.wheel_dir:
-        os.environ["PREBUILT_WHEEL_DIR"] = args.wheel_dir
-    else:
-        os.environ["PREBUILT_WHEEL_DIR"] = os.path.join(root_dir, ".wheels")
-
-    if in_ci():
-        # prepare a build of eng/tools/azure-sdk-tools
-        build_whl_for_req("eng/tools/azure-sdk-tools", root_dir, os.environ.get("PREBUILT_WHEEL_DIR"))
-
-    # so if we have checks whl,import_all and selected package paths `sdk/core/azure-core`, `sdk/storage/azure-storage-blob` we should
-    # shell out to `azypysdk <checkname>` with cwd of the package directory, which is what is in `targeted_packages` array
-    # each individual thread may need to re-invoke if they need to self-isolate themselves, but we don't have to worry about that.
-
-    # Prepare check list
-    raw_checks = (args.checks_list or "").split(",")
-    checks = [c.strip() for c in raw_checks if c and c.strip()]
-    if not checks:
-        logging.error("No valid checks provided via -c/--checks.")
-        sys.exit(2)
-
-    logging.info(f"Running {len(checks)} checks across {len(targeted_packages)} packages (max_parallel={args.max_parallel}).")
-
-    configure_interrupt_handling()
-    try:
-        exit_code = asyncio.run(run_all_checks(targeted_packages, checks, args.max_parallel))
-    except KeyboardInterrupt:
-        logging.error("Aborted by user.")
-        exit_code = 130
-    sys.exit(exit_code)
-
diff --git a/eng/tools/azure-sdk-tools/azpysdk/Check.py b/eng/tools/azure-sdk-tools/azpysdk/Check.py
index 0382b0263cc2..22b61665640f 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/Check.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/Check.py
@@ -4,7 +4,7 @@
 import traceback
 import sys
 
-from typing import Sequence, Optional, List, Any
+from typing import Sequence, Optional, List, Any, Tuple
 from subprocess import check_call
 
 from ci_tools.parsing import ParsedSetup
@@ -45,11 +45,9 @@ def run(self, args: argparse.Namespace) -> int:
         """
         return 0
 
-    def handle_venv(self, isolate: bool, args: argparse.Namespace, venv_location: Optional[str]) -> str:
-        """Handle virtual environment commands."""
-        # if we have to isolate, return the new python exe that the rest of the checks should use
+    def create_venv(self, isolate: bool,venv_location: Optional[str]) -> str:
+        """Abstraction for creating a virtual environment."""
         if (isolate):
-            # os.environ["AZURE_SDK_TOOLS_VENV"] = "1"
 
             venv_cmd = get_venv_call()
             if not venv_location:
@@ -76,6 +74,18 @@ def handle_venv(self, isolate: bool, args: argparse.Namespace, venv_location: Op
         # if we don't need to isolate, just return the python executable that we're invoking
         return sys.executable
 
+    def get_executable(self, isolate: bool, check_name: str, executable: str, package_folder: str) -> Tuple[str, str]:
+        """Get the Python executable that should be used for this check."""
+        venv_location = os.path.join(package_folder, f".venv_{check_name}")
+
+        # if isolation is required, the executable we get back will align with the venv
+        # otherwise we'll just get sys.executable and install in current
+        executable = self.create_venv(isolate, venv_location)
+        staging_directory = os.path.join(venv_location, ".staging")
+        os.makedirs(staging_directory, exist_ok=True)
+        return executable, staging_directory
+
+
     def get_targeted_directories(self, args: argparse.Namespace) -> List[ParsedSetup]:
         """
         Get the directories that are targeted for the check.
diff --git a/eng/tools/azure-sdk-tools/azpysdk/import_all.py b/eng/tools/azure-sdk-tools/azpysdk/import_all.py
index 0dd4ea27cefd..cd6750eefb7d 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/import_all.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/import_all.py
@@ -53,18 +53,18 @@ def run(self, args: argparse.Namespace) -> int:
 
         for parsed in targeted:
             pkg = parsed.folder
-                
+            executable, staging_directory = self.get_executable(args.isolate, args.command, sys.executable, pkg)
 
-            staging_area = tempfile.mkdtemp()
             create_package_and_install(
-                distribution_directory=staging_area,
+                distribution_directory=staging_directory,
                 target_setup=pkg,
                 skip_install=False,
                 cache_dir=None,
-                work_dir=staging_area,
+                work_dir=staging_directory,
                 force_create=False,
                 package_type="wheel",
                 pre_download_disabled=False,
+                python_executable=executable
             )
 
             if should_run_import_all(parsed.name):
@@ -76,7 +76,7 @@ def run(self, args: argparse.Namespace) -> int:
                 )
                 import_script_all = "from {0} import *".format(parsed.namespace)
                 commands = [
-                    sys.executable,
+                    executable,
                     "-c",
                     import_script_all
                 ]
diff --git a/eng/tools/azure-sdk-tools/azpysdk/mypy.py b/eng/tools/azure-sdk-tools/azpysdk/mypy.py
index 7d0e0ae4256c..255266ab3a74 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/mypy.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/mypy.py
@@ -9,7 +9,7 @@
 
 from .Check import Check
 from ci_tools.parsing import ParsedSetup
-from ci_tools.functions import discover_targeted_packages
+from ci_tools.functions import pip_install
 from ci_tools.scenario.generation import create_package_and_install
 from ci_tools.variables import in_ci, set_envvar_defaults
 from ci_tools.environment_exclusions import (
@@ -33,8 +33,8 @@ def register(self, subparsers: "argparse._SubParsersAction", parent_parsers: Opt
         p.set_defaults(func=self.run)
 
         p.add_argument(
-            "--next", 
-            default=False, 
+            "--next",
+            default=False,
             help="Next version of mypy is being tested",
             required=False
         )
@@ -52,27 +52,27 @@ def run(self, args: argparse.Namespace) -> int:
         for parsed in targeted:
             package_dir = parsed.folder
             package_name = parsed.name
+            executable, staging_directory = self.get_executable(args.isolate, args.command, sys.executable, pkg)
             print(f"Processing {package_name} for mypy check")
-
-            staging_area = tempfile.mkdtemp()
             create_package_and_install(
-                distribution_directory=staging_area,
+                distribution_directory=staging_directory,
                 target_setup=package_dir,
                 skip_install=False,
                 cache_dir=None,
-                work_dir=staging_area,
+                work_dir=staging_directory,
                 force_create=False,
                 package_type="wheel",
                 pre_download_disabled=False,
+                python_executable=executable
             )
 
             # install mypy
             try:
                 if (args.next):
                     # use latest version of mypy
-                    check_call([sys.executable, "-m", "pip", "install", "mypy"])
+                    pip_install(["mypy"], True, executable, package_dir)
                 else:
-                    check_call([sys.executable, "-m", "pip", "install", f"mypy=={MYPY_VERSION}"])
+                    pip_install([f"mypy=={MYPY_VERSION}"], True, executable, package_dir)
             except CalledProcessError as e:
                 print("Failed to install mypy:", e)
                 return e.returncode
@@ -89,7 +89,7 @@ def run(self, args: argparse.Namespace) -> int:
             top_level_module = parsed.namespace.split(".")[0]
 
             commands = [
-                sys.executable,
+                executable,
                 "-m",
                 "mypy",
                 "--python-version",
@@ -107,9 +107,9 @@ def run(self, args: argparse.Namespace) -> int:
                 results.append(check_call(src_code))
                 logging.info("Verified mypy, no issues found")
             except CalledProcessError as src_error:
-                src_code_error = src_error 
+                src_code_error = src_error
                 results.append(src_error.returncode)
-            
+
             if not args.next and in_ci() and not is_check_enabled(package_dir, "type_check_samples", True):
                 logging.info(
                     f"Package {package_name} opts-out of mypy check on samples."
@@ -145,6 +145,5 @@ def run(self, args: argparse.Namespace) -> int:
                     create_vnext_issue(package_dir, "mypy")
                 else:
                     close_vnext_issue(package_name, "mypy")
-            
+
         return max(results) if results else 0
-        
\ No newline at end of file
diff --git a/eng/tools/azure-sdk-tools/azpysdk/whl.py b/eng/tools/azure-sdk-tools/azpysdk/whl.py
index d5d2d127772a..69acc3a62994 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/whl.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/whl.py
@@ -8,7 +8,7 @@
 
 from .Check import Check
 
-from ci_tools.functions import discover_targeted_packages, is_error_code_5_allowed, pip_install
+from ci_tools.functions import is_error_code_5_allowed, pip_install
 from ci_tools.variables import set_envvar_defaults
 from ci_tools.parsing import ParsedSetup
 from ci_tools.scenario.generation import create_package_and_install
@@ -39,12 +39,7 @@ def run(self, args: argparse.Namespace) -> int:
 
         for parsed in targeted:
             pkg = parsed.folder
-            venv_location = os.path.join(parsed.folder, f".venv_{args.command}")
-            # if isolation is required, the executable we get back will align with the venv
-            # otherwise we'll just get sys.executable and install in current
-            executable = self.handle_venv(args.isolate, args, venv_location=venv_location)
-            staging_directory = os.path.join(venv_location, ".staging")
-            os.makedirs(staging_directory, exist_ok=True)
+            executable, staging_directory = self.get_executable(args.isolate, args.command, sys.executable, pkg)
 
             print(f"Invoking check with {executable}")
             dev_requirements = os.path.join(pkg, "dev_requirements.txt")

From 24f1df92c14c14372fb449b406999289e0159924 Mon Sep 17 00:00:00 2001
From: Scott Beddall <scbedd@microsoft.com>
Date: Thu, 28 Aug 2025 20:56:07 +0000
Subject: [PATCH 09/12] updates to venv isolation

---
 eng/tools/azure-sdk-tools/README.md | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/eng/tools/azure-sdk-tools/README.md b/eng/tools/azure-sdk-tools/README.md
index f6dde94c6521..47bfa5748a61 100644
--- a/eng/tools/azure-sdk-tools/README.md
+++ b/eng/tools/azure-sdk-tools/README.md
@@ -217,6 +217,9 @@ class my_check(Check):
         for parsed in targeted:
             pkg_dir = parsed.folder
             pkg_name = parsed.name
+            executable, staging_directory = self.get_executable(args.isolate, args.command, sys.executable, pkg)
+            # the rest of the check should use executable, not sys.executable
+            # if a staging area is needed use staging_directory
             print(f"Processing {pkg_name} for my_check")
 ```
 

From 5c2878ef1959e0d129be1cd767a9940ce84001bf Mon Sep 17 00:00:00 2001
From: Scott Beddall <scbedd@microsoft.com>
Date: Thu, 28 Aug 2025 21:02:17 +0000
Subject: [PATCH 10/12] remove extraneous comments

---
 eng/tools/azure-sdk-tools/azpysdk/Check.py | 17 ++---------------
 1 file changed, 2 insertions(+), 15 deletions(-)

diff --git a/eng/tools/azure-sdk-tools/azpysdk/Check.py b/eng/tools/azure-sdk-tools/azpysdk/Check.py
index 22b61665640f..6fd988b8b359 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/Check.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/Check.py
@@ -45,20 +45,10 @@ def run(self, args: argparse.Namespace) -> int:
         """
         return 0
 
-    def create_venv(self, isolate: bool,venv_location: Optional[str]) -> str:
+    def create_venv(self, isolate: bool, venv_location: str) -> str:
         """Abstraction for creating a virtual environment."""
         if (isolate):
-
-            venv_cmd = get_venv_call()
-            if not venv_location:
-                venv_location = os.path.join(REPO_ROOT, f".venv_{args.command}")
-
-            # todo, make this a consistent directory based on the command
-            # I'm seriously thinking we should move handle_venv within each check's main(),
-            # which will mean that we will _know_ what folder we're in.
-            # however, that comes at the cost of not having every check be able to handle one or multiple packages
-            # I don't want to get into an isolation loop where every time we need a new venv, we create it, call it,
-            # and now as we foreach across the targeted packages we've lost our spot.
+            venv_cmd = get_venv_call(sys.executable)
             check_call(venv_cmd + [venv_location])
 
             # TODO: we should reuse part of build_whl_for_req to integrate with PREBUILT_WHL_DIR so that we don't have to fresh build for each
@@ -66,10 +56,7 @@ def create_venv(self, isolate: bool,venv_location: Optional[str]) -> str:
             install_into_venv(venv_location, os.path.join(REPO_ROOT, "eng/tools/azure-sdk-tools"), False, "build")
             venv_python_exe = get_venv_python(venv_location)
 
-            # here is a newly prepped virtual environment (which includes azure-sdk-tools)
             return venv_python_exe
-            # command_args = [venv_python_exe, "-m", "azpysdk.main"] + sys.argv[1:]
-            # check_call(command_args)
 
         # if we don't need to isolate, just return the python executable that we're invoking
         return sys.executable

From aea17d7660611120dac634cf461db552f1349625 Mon Sep 17 00:00:00 2001
From: Scott Beddall <scbedd@microsoft.com>
Date: Thu, 28 Aug 2025 21:50:06 +0000
Subject: [PATCH 11/12] fixed up the mypy invocation

---
 eng/tools/azure-sdk-tools/azpysdk/mypy.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/eng/tools/azure-sdk-tools/azpysdk/mypy.py b/eng/tools/azure-sdk-tools/azpysdk/mypy.py
index 255266ab3a74..a6641b6c21cd 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/mypy.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/mypy.py
@@ -52,7 +52,7 @@ def run(self, args: argparse.Namespace) -> int:
         for parsed in targeted:
             package_dir = parsed.folder
             package_name = parsed.name
-            executable, staging_directory = self.get_executable(args.isolate, args.command, sys.executable, pkg)
+            executable, staging_directory = self.get_executable(args.isolate, args.command, sys.executable, package_dir)
             print(f"Processing {package_name} for mypy check")
             create_package_and_install(
                 distribution_directory=staging_directory,

From 3027631adb0b00314cf50f8025c29082d6be742f Mon Sep 17 00:00:00 2001
From: Scott Beddall <scbedd@microsoft.com>
Date: Thu, 28 Aug 2025 21:51:11 +0000
Subject: [PATCH 12/12] change to use run instead of check_call

---
 eng/tools/azure-sdk-tools/azpysdk/whl.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/eng/tools/azure-sdk-tools/azpysdk/whl.py b/eng/tools/azure-sdk-tools/azpysdk/whl.py
index 69acc3a62994..a1cc2e8af5cb 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/whl.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/whl.py
@@ -4,7 +4,7 @@
 import os
 from typing import Optional, List, Any
 import sys
-from subprocess import check_call
+from subprocess import run
 
 from .Check import Check
 
@@ -65,7 +65,7 @@ def run(self, args: argparse.Namespace) -> int:
             # TODO: split sys.argv[1:] on -- and pass in everything after the -- as additional arguments
             # TODO: handle mark_args
             logging.info(f"Invoke pytest for {pkg}")
-            exit_code = check_call(
+            exit_code = run(
                 [executable, "-m", "pytest", "."] + [
                     "-rsfE",
                     f"--junitxml={pkg}/test-junit-{args.command}.xml",
@@ -79,7 +79,7 @@ def run(self, args: argparse.Namespace) -> int:
                     "--ignore=samples"
                 ]
                 , cwd=pkg
-            )
+            ).returncode
 
             if exit_code != 0:
                 if exit_code == 5 and is_error_code_5_allowed(parsed.folder, parsed.name):