diff --git a/.azure-devops/merge.yml b/.azure-devops/merge.yml
index 7c5206a02..30bc8bd27 100644
--- a/.azure-devops/merge.yml
+++ b/.azure-devops/merge.yml
@@ -48,71 +48,6 @@ jobs:
       - template: templates/setup-python.yml
       - template: templates/build-publish-azure-cli-test-sdk.yml
 
-  - job: "run_unit_tests_ubuntu"
-    dependsOn:
-      [
-        "build_and_publish_azure_iot_cli_ext",
-        "build_and_publish_azure_cli_test_sdk",
-      ]
-    pool:
-      vmImage: ${{ parameters.linuxImage }}
-    strategy:
-      matrix:
-        Python310:
-          python.version: "3.10"
-        Python311:
-          python.version: "3.11"
-        Python312:
-          python.version: "3.12"
-        Python313:
-          python.version: "3.13"
-
-    steps:
-      - bash: sudo rm -R -f /usr/local/lib/azureExtensionDir
-
-      - template: templates/run-tests-parallel.yml
-        parameters:
-          pythonVersion: "$(python.version)"
-          runUnitTests: true
-          runIntTests: false
-
-  - job: "run_unit_tests_macOs"
-    dependsOn:
-      [
-        "build_and_publish_azure_iot_cli_ext",
-        "build_and_publish_azure_cli_test_sdk",
-      ]
-    pool:
-      vmImage: "macOS-15"
-
-    steps:
-      - template: templates/run-tests-parallel.yml
-        parameters:
-          pythonVersion: "3.12"
-          runUnitTests: true
-          runIntTests: false
-
-  - job: "run_unit_tests_windows"
-    dependsOn:
-      [
-        "build_and_publish_azure_iot_cli_ext",
-        "build_and_publish_azure_cli_test_sdk",
-      ]
-    pool:
-      vmImage: ${{ parameters.windowsImage }}
-
-    steps:
-      - task: PowerShell@2
-        inputs:
-          targetType: "inline"
-          script: 'ren "C:\Program Files\Common Files\AzureCliExtensionDirectory" "C:\Program Files\Common Files\AzureCliExtensionDirectory1"'
-
-      - template: templates/run-tests-parallel.yml
-        parameters:
-          pythonVersion: "3.12"
-          runUnitTests: true
-          runIntTests: false
-
   - job: "run_style_check"
     dependsOn:
       [
diff --git a/.github/workflows/ci_workflow.yml b/.github/workflows/ci_workflow.yml
index 2872ce62a..76d02665e 100644
--- a/.github/workflows/ci_workflow.yml
+++ b/.github/workflows/ci_workflow.yml
@@ -8,12 +8,21 @@ concurrency:
 on:
   pull_request:
   push:
+    branches:
+      - dev
+      - preview
+      - 1.1.0-preview
+      - "release/**"
+    tags:
+      - "**"
   workflow_dispatch:
 jobs:
   build:
     uses: ./.github/workflows/ci_build.yml
   test:
     uses: ./.github/workflows/tox.yml
+    with:
+      pr-matrix: ${{ github.event_name == 'pull_request' }}
   linter:
     needs: [build]
     uses: ./.github/workflows/azdev_linter.yml
diff --git a/.github/workflows/int_test.yml b/.github/workflows/int_test.yml
index a87737a19..63fc87f12 100644
--- a/.github/workflows/int_test.yml
+++ b/.github/workflows/int_test.yml
@@ -277,8 +277,10 @@ jobs:
           echo "|---------|--------|--------|--------------|---------|" >> $GITHUB_STEP_SUMMARY
           echo "$matrix" | jq -r '.[] | "| \(.service) | \(.python) | \(.region) | \(.arm_endpoint) | \(.timeout) min |"' >> $GITHUB_STEP_SUMMARY
 
-  unit-test:
-    name: "Linter and unit tests"
+  lint:
+    name: "Linter"
+    permissions:
+      contents: read
     runs-on: ubuntu-latest
     timeout-minutes: 25
     steps:
@@ -289,23 +291,104 @@ jobs:
         uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
         with:
           python-version: "3.13"
-          cache: "pip" # Cache pip downloads to avoid re-downloading packages across runs
+          cache: "pip"
 
-      - name: "Setup test suite"
+      - name: "Run linter"
         run: |
           python -m pip install tox
-          tox r -vv --notest
+          python -m tox r -e lint -vv
 
-      - name: "Run linter and unit tests"
-        run: tox r --skip-pkg-install
+  unit-shards:
+    name: "Unit tests (shard ${{ matrix.shard }}/4)"
+    permissions:
+      contents: read
+    runs-on: ubuntu-latest
+    timeout-minutes: 25
+    strategy:
+      fail-fast: false
+      max-parallel: 4
+      matrix:
+        shard: [1, 2, 3, 4]
+    steps:
+      - name: "Checkout source"
+        uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5.1.0
 
-      - name: "Upload coverage report"
+      - name: "Setup Python 3.13"
+        uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
+        with:
+          python-version: "3.13"
+          cache: "pip"
+
+      - name: "Run balanced serial unit shard"
+        env:
+          SHARD: ${{ matrix.shard }}
+          UNIT_RUN_ID: "github:${{ github.repository }}:${{ github.run_id }}"
+          UNIT_COMMIT: ${{ github.sha }}
+          UNIT_ATTEMPT: ${{ github.run_attempt }}
+          RESULTS: ${{ github.workspace }}/unit-result
+          COVERAGE_FILE: ${{ github.workspace }}/unit-result/coverage.dat
+        run: |
+          python -m pip install tox
+          python -m tox r -e python-azcur-unit -vv -- \
+            -p azext_iot.tests._unit_shard_plugin --unit-shard "$SHARD" \
+            --unit-shard-output "$RESULTS" --junitxml "$RESULTS/junit.xml"
+
+      - name: "Upload unit shard evidence"
         if: ${{ always() }}
+        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
+        with:
+          name: unit-shard-${{ matrix.shard }}-${{ github.run_attempt }}
+          path: unit-result/
+          if-no-files-found: error
+          retention-days: 30
+
+  unit-test:
+    name: "Linter and unit test gate"
+    needs: [lint, unit-shards]
+    permissions:
+      contents: read
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - name: "Checkout source"
+        uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5.1.0
+
+      - name: "Setup Python 3.13"
+        uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
+        with:
+          python-version: "3.13"
+
+      - name: "Download unit shard evidence"
+        uses: actions/download-artifact@37930b1c2abaa49bbe596cd826c3c89aef350131 # v7.0.0
+        with:
+          pattern: unit-shard-*
+          path: unit-history
+
+      - name: "Require complete unit coverage and combine results"
+        env:
+          UNIT_RUN_ID: "github:${{ github.repository }}:${{ github.run_id }}"
+          UNIT_COMMIT: ${{ github.sha }}
+        run: |
+          python -m pip install coverage
+          python azext_iot/tests/_unit_shards.py --history unit-history --output unit-coverage
+
+      - name: "Upload unit timing and completeness reports"
+        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
+        with:
+          name: unit-summary-${{ github.run_attempt }}
+          path: |
+            unit-coverage/summary.json
+            unit-coverage/timings.json
+          if-no-files-found: error
+          retention-days: 30
+
+      - name: "Upload coverage report"
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
         with:
           name: coverage-unit
           overwrite: true
-          path: ./.coverage
+          path: unit-coverage/.coverage
+          if-no-files-found: error
           include-hidden-files: true
           retention-days: 30
 
diff --git a/.github/workflows/tox.yml b/.github/workflows/tox.yml
index 3c1a198d6..7eb1cb29a 100644
--- a/.github/workflows/tox.yml
+++ b/.github/workflows/tox.yml
@@ -2,6 +2,11 @@ name: Tox tests
 on:
   workflow_call:
     inputs:
+      pr-matrix:
+        description: Test Python 3.13 on every OS and Python 3.10 on Ubuntu only
+        type: boolean
+        required: false
+        default: false
       continue-on-error:
         type: boolean
         required: false
@@ -19,7 +24,7 @@ permissions:
 
 jobs:
   tox:
-    name: Unit test ${{ matrix.py }} - ${{ matrix.os }}
+    name: Unit test ${{ matrix.py }} - ${{ matrix.os }} (shard ${{ matrix.shard }}/4)
     continue-on-error: ${{ inputs.continue-on-error }}
     runs-on: ${{ matrix.os }}
     strategy:
@@ -34,6 +39,8 @@ jobs:
           - "3.12"
           - "3.11"
           - "3.10"
+        shard: [1, 2, 3, 4]
+        exclude: ${{ fromJSON(inputs.pr-matrix && '[{"py":"3.12"},{"py":"3.11"},{"os":"windows-2025","py":"3.10"},{"os":"macos-15-intel","py":"3.10"}]' || '[]') }}
     steps:
       - name: Setup python ${{ matrix.py }}
         uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
@@ -49,9 +56,85 @@ jobs:
       - name: Setup test suite
         run: |
           python -m pip install tox
-          tox r -vv --notest
+          python -m tox r -e python-azcur-unit -vv --notest
       - name: Run test suite
-        run: tox r --skip-pkg-install
+        env:
+          UNIT_RUN_ID: "github:${{ github.repository }}:${{ github.run_id }}:tox:${{ matrix.os }}:py${{ matrix.py }}"
+          UNIT_COMMIT: ${{ github.sha }}
+          UNIT_ATTEMPT: ${{ github.run_attempt }}
+          COVERAGE_FILE: ${{ github.workspace }}/unit-result/coverage.dat
+        run: >-
+          python -m tox r -e python-azcur-unit --skip-pkg-install --
+          -p azext_iot.tests._unit_shard_plugin --unit-shard ${{ matrix.shard }}
+          --unit-shard-output unit-result --junitxml unit-result/junit.xml
+      - name: Upload unit shard evidence
+        if: ${{ always() }}
+        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
+        with:
+          name: tox-unit-${{ matrix.os }}-py${{ matrix.py }}-${{ matrix.shard }}-${{ github.run_attempt }}
+          path: unit-result/
+          if-no-files-found: error
+          retention-days: 30
+      - name: Run lint once per OS and Python
+        if: ${{ matrix.shard == 1 }}
+        run: python -m tox r -e lint -vv
+
+  unit-gate:
+    name: Unit test ${{ matrix.py }} - ${{ matrix.os }}
+    needs: tox
+    if: ${{ always() }}
+    continue-on-error: ${{ inputs.continue-on-error }}
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        os:
+          - ubuntu-24.04
+          - windows-2025
+          - macos-15-intel
+        py:
+          - "3.13"
+          - "3.12"
+          - "3.11"
+          - "3.10"
+        exclude: ${{ fromJSON(inputs.pr-matrix && '[{"py":"3.12"},{"py":"3.11"},{"os":"windows-2025","py":"3.10"},{"os":"macos-15-intel","py":"3.10"}]' || '[]') }}
+    steps:
+      - name: Require successful unit shards and lint
+        if: ${{ needs.tox.result != 'success' }}
+        run: |
+          echo "::error::Unit shards or lint did not succeed (${{ needs.tox.result }})."
+          exit 1
+      - name: Setup python ${{ matrix.py }}
+        uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
+        with:
+          python-version: ${{ matrix.py }}
+      - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5.1.0
+      - name: Download this matrix combination's unit evidence
+        uses: actions/download-artifact@37930b1c2abaa49bbe596cd826c3c89aef350131 # v7.0.0
+        with:
+          pattern: tox-unit-${{ matrix.os }}-py${{ matrix.py }}-*
+          path: unit-history
+      - name: Install coverage
+        run: python -m pip install coverage
+      - name: Verify complete unit execution and combine coverage
+        env:
+          UNIT_RUN_ID: "github:${{ github.repository }}:${{ github.run_id }}:tox:${{ matrix.os }}:py${{ matrix.py }}"
+          UNIT_COMMIT: ${{ github.sha }}
+        run: >-
+          python azext_iot/tests/_unit_shards.py --history unit-history --output unit-coverage
+          --prefix tox-unit-${{ matrix.os }}-py${{ matrix.py }}
+      - name: Generate coverage reports
+        env:
+          COVERAGE_FILE: ${{ github.workspace }}/unit-coverage/.coverage
+        run: python -m coverage report && python -m coverage html && python -m coverage json
+      - name: Upload verified unit results
+        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
+        with:
+          name: tox-summary-${{ matrix.os }}-py${{ matrix.py }}-${{ github.run_attempt }}
+          path: unit-coverage/
+          include-hidden-files: true
+          if-no-files-found: error
+          retention-days: 30
       - name: Upload coverage report
         if: ${{ matrix.os == 'ubuntu-24.04' && matrix.py == '3.13' }}
         continue-on-error: true
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 75fa80ad8..2c721e036 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -137,6 +137,46 @@ Execute the following command to run the IoT Hub unit tests:
 
 `pytest azext_iot/tests/iothub/ -k "_unit.py"`
 
+#### Parallel CI unit tests
+
+The GitHub integration precheck uses four balanced unit-test shards on
+Linux/Python 3.13. PR CI uses the same runner on Python 3.13 across Ubuntu,
+Windows, and macOS, plus Python 3.10 on Ubuntu: four combinations and 16 unit jobs.
+GitHub branch/tag pushes, manual runs, and release workflows retain all three operating
+systems and Python 3.10-3.13: twelve combinations and 48 unit jobs. The reusable
+workflow defaults to the full matrix; only the CI caller's `pull_request` event
+enables `pr-matrix`. Each combination runs all unit tests across four shards,
+with lint once per combination and a coverage gate for each. Each shard uses
+serial pytest, not xdist, and new unit files are included automatically.
+Shards must use the same Python major/minor version; runner patch versions may differ.
+PR aggregate gates retain the `test / Unit test <python> - <os>` names required
+by branch rules and fail if any shard or lint run fails.
+
+Feature-branch updates run GitHub CI through `pull_request`, without a duplicate
+push-triggered matrix. Push CI remains enabled for `dev`, `preview`,
+`1.1.0-preview`, `release/**`, and tags; manual dispatch is unchanged.
+
+Azure Pipelines Merge (pipeline 11) no longer repeats the unit suite. It retains
+wheel and software-manifest generation, the Azure CLI test SDK build, style and
+command-table lint, and Credential Scan. The four GitHub PR combinations are
+the PR unit gate; the former additional ADO unit combinations and tests against
+the unreleased CLI test SDK are no longer part of Merge validation. Full-matrix
+GitHub runs and other integration/unit pipelines are unchanged.
+
+Integration starts only after lint, all shards, and the `unit-test` gate pass.
+The gate checks complete, exactly-once case coverage and combines coverage into
+`coverage-unit`. Native reruns can reuse successful shards from the same run and
+commit, but incomplete newer attempts cannot fall back to older results.
+
+Shard results are published as `unit-shard-<number>-<attempt>`.
+PR CI uses `tox-unit-<os>-py<version>-<number>-<attempt>` to keep each combination
+separate from the others and from integration prechecks.
+Integration summaries use `unit-summary-<attempt>`; PR summaries use
+`tox-summary-<os>-py<version>-<attempt>`. Both include the completeness summary
+and `timings.json`; use those timings to maintain `azext_iot/tests/unit_test_durations.json`.
+The shared runner accepts CI-neutral run, commit, and attempt inputs so Azure
+Pipelines can use the same implementation.
+
 ### Integration Tests
 
 Integration tests are run against Azure resources and depend on environment variables.
diff --git a/azext_iot/tests/_unit_shard_plugin.py b/azext_iot/tests/_unit_shard_plugin.py
new file mode 100644
index 000000000..cba668c02
--- /dev/null
+++ b/azext_iot/tests/_unit_shard_plugin.py
@@ -0,0 +1,92 @@
+# coding=utf-8
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+"""Collect the complete unit inventory, then execute one deterministic file shard."""
+
+from collections import Counter
+import os
+from pathlib import Path
+import platform
+import time
+
+import pytest
+
+from azext_iot.tests import _unit_shards as shards
+
+
+def pytest_addoption(parser):
+    parser.addoption("--unit-shard", type=int, choices=range(1, shards.COUNT + 1))
+    parser.addoption("--unit-shard-output")
+
+
+def pytest_configure(config):
+    if not config.getoption("unit_shard") or not config.getoption("unit_shard_output"):
+        raise pytest.UsageError("Unit sharding requires a shard number and a unique output directory.")
+    if (config.getoption("numprocesses", None) not in (None, 0, "0") or config.getoption("reruns", 0)
+            or config.getoption("keyword") != "_unit.py" or config.getoption("markexpr")
+            or config.getoption("deselect", []) or config.getoption("lf", False)):
+        raise pytest.UsageError("Unit shards require serial pytest, the complete _unit.py selection and no retries.")
+    if [Path(arg).resolve() for arg in config.args] != [config.rootpath / "azext_iot/tests"]:
+        raise pytest.UsageError("Unit shards must collect the complete azext_iot/tests directory.")
+    config.pluginmanager.register(Receipt(config), "unit-shard-receipt")
+
+
+class Receipt:
+    def __init__(self, config):
+        self.output = Path(config.getoption("unit_shard_output")).resolve()
+        self.output.mkdir(parents=True, exist_ok=True)
+        self.started = time.monotonic()
+        self.profile = shards.read(shards.PROFILE)
+        self.identities = {}
+        self.data = {
+            "schema": 1, "context": shards.context(), "python": platform.python_version(),
+            "shard": config.getoption("unit_shard"), "attempt": int(os.environ["UNIT_ATTEMPT"]),
+            "profile": shards.digest(self.profile), "inventory": [], "selected": [], "reports": {},
+            "durations": {}, "finished": False, "exitstatus": None, "artifacts": {},
+        }
+        shards.write(self.output / "receipt.json", self.data, exclusive=True)
+
+    @pytest.hookimpl(trylast=True)
+    def pytest_collection_modifyitems(self, config, items):
+        counts = Counter()
+        for item in items:
+            if not item.path.name.endswith("_unit.py"):
+                raise pytest.UsageError("Non-unit cases matched the unit selection.")
+            # Random parameter values differ between processes. Source function and
+            # parameter position retain every case without persisting those values.
+            base = item.nodeid.split("[", 1)[0]
+            self.identities[item.nodeid] = f"{base}::{counts[base]}"
+            counts[base] += 1
+        inventory = sorted(self.identities.values())
+        if len(inventory) != len(items):
+            raise pytest.UsageError("Duplicate unit collection identities.")
+        self.data["inventory"] = inventory
+        plan = shards.partition([node.split("::", 1)[0] for node in inventory], self.profile)
+        selected, dropped = [], []
+        for item in items:
+            name = item.nodeid.split("::", 1)[0]
+            (selected if name in plan[self.data["shard"] - 1] else dropped).append(item)
+        self.data["selected"] = sorted(self.identities[item.nodeid] for item in selected)
+        items[:] = selected
+        config.hook.pytest_deselected(items=dropped)
+        shards.write(self.output / "receipt.json", self.data)
+
+    def pytest_runtest_logreport(self, report):
+        node = self.identities[report.nodeid]
+        self.data["reports"].setdefault(node, {}).setdefault(report.when, []).append(report.outcome)
+        name = node.split("::", 1)[0]
+        self.data["durations"][name] = self.data["durations"].get(name, 0) + report.duration
+
+    @pytest.hookimpl(hookwrapper=True, tryfirst=True)
+    def pytest_sessionfinish(self, session, exitstatus):
+        outcome = yield
+        self.data["finished"] = outcome.excinfo is None
+        self.data["exitstatus"] = int(session.exitstatus)
+        self.data["seconds"] = time.monotonic() - self.started
+        self.data["artifacts"] = {
+            name: shards.file_digest(self.output / name) for name in shards.ARTIFACTS if (self.output / name).is_file()
+        }
+        shards.write(self.output / "receipt.json", self.data)
diff --git a/azext_iot/tests/_unit_shards.py b/azext_iot/tests/_unit_shards.py
new file mode 100644
index 000000000..f3cd08191
--- /dev/null
+++ b/azext_iot/tests/_unit_shards.py
@@ -0,0 +1,138 @@
+# coding=utf-8
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+"""Deterministic unit-file balancing and fail-closed aggregation of four serial pytest jobs."""
+
+import argparse
+import hashlib
+import json
+import math
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+
+COUNT = 4
+PROFILE = Path(__file__).with_name("unit_test_durations.json")
+ARTIFACTS = ("coverage.dat", "junit.xml")
+
+
+def read(path):
+    return json.loads(Path(path).read_text(encoding="utf-8"))
+
+
+def write(path, value, *, exclusive=False):
+    with Path(path).open("x" if exclusive else "w", encoding="utf-8") as stream:
+        stream.write(json.dumps(value, indent=2, sort_keys=True) + "\n")
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+
+def file_digest(path):
+    return hashlib.sha256(Path(path).read_bytes()).hexdigest()
+
+
+def partition(files, profile):
+    files = sorted(set(files))
+    weights = {name: profile["seconds"].get(name, profile["default_seconds"]) for name in files}
+    if len(files) < COUNT or any(not isinstance(w, (int, float)) or not math.isfinite(w) or w <= 0 for w in weights.values()):
+        raise ValueError("Four nonempty shards require valid positive file-duration weights.")
+    loads, shards = [0.0] * COUNT, [[] for _ in range(COUNT)]
+    for name in sorted(files, key=lambda name: (-weights[name], name)):
+        index = min(range(COUNT), key=lambda index: (loads[index], index))
+        shards[index].append(name)
+        loads[index] += weights[name]
+    return [sorted(shard) for shard in shards]
+
+
+def context():
+    return {"build": os.environ["UNIT_RUN_ID"], "commit": os.environ["UNIT_COMMIT"]}
+
+
+def python_series(version):
+    match = re.fullmatch(r"([0-9]+\.[0-9]+)\.[0-9]+", version) if isinstance(version, str) else None
+    if match is None:
+        raise ValueError("Unit shard Python version must have a major, minor and patch number.")
+    return match.group(1)
+
+
+def validate(records, expected_context, profile):
+    if len(records) != COUNT or {r["shard"] for r in records} != set(range(1, COUNT + 1)):
+        raise ValueError("Exactly four distinct unit shards are required.")
+    if len({python_series(record["python"]) for record in records}) != 1:
+        raise ValueError("Unit shards must use the same Python major/minor version.")
+    inventory = records[0]["inventory"]
+    if not inventory or inventory != sorted(set(inventory)):
+        raise ValueError("Full unit collection is empty or duplicated.")
+    plan = partition([node.split("::", 1)[0] for node in inventory], profile)
+    selected = []
+    for record in records:
+        expected = [node for node in inventory if node.split("::", 1)[0] in plan[record["shard"] - 1]]
+        if (record.get("schema") != 1 or record["context"] != expected_context
+                or record["profile"] != digest(profile)
+                or record["inventory"] != inventory or record["selected"] != expected
+                or record["finished"] is not True or record["exitstatus"] != 0
+                or set(record["reports"]) != set(expected)):
+            raise ValueError("Unit shard identity, collection, selection or terminal execution is invalid.")
+        for stages in record["reports"].values():
+            setup_skip = stages.get("setup") == ["skipped"]
+            if (stages.get("teardown") != ["passed"]
+                    or set(stages) != ({"setup", "teardown"} if setup_skip else {"setup", "call", "teardown"})
+                    or not setup_skip and (stages.get("setup") != ["passed"]
+                                           or stages.get("call") not in (["passed"], ["skipped"]))):
+                raise ValueError("A unit case failed or has missing/duplicate execution stages.")
+        selected.extend(record["selected"])
+    if sorted(selected) != inventory:
+        raise ValueError("Unit shards omit or duplicate cases.")
+    return len(inventory)
+
+
+def aggregate(history, output, prefix="unit-shard"):
+    latest = {}
+    for folder in Path(history).glob(f"{prefix}-*"):
+        match = re.fullmatch(rf"{re.escape(prefix)}-([1-4])-([1-9][0-9]*)", folder.name)
+        if not match or not folder.is_dir():
+            raise ValueError("Unexpected unit-shard artifact.")
+        shard, attempt = map(int, match.groups())
+        if shard not in latest or attempt > latest[shard][0]:
+            latest[shard] = (attempt, folder)
+    records, coverage = [], []
+    for shard, (attempt, folder) in sorted(latest.items()):
+        record = read(folder / "receipt.json")
+        if record["shard"] != shard or record["attempt"] != attempt:
+            raise ValueError("Unit artifact name disagrees with its receipt.")
+        if record.get("artifacts") != {name: file_digest(folder / name) for name in ARTIFACTS}:
+            raise ValueError("Unit results/coverage are missing or changed.")
+        records.append(record)
+        coverage.append(str((folder / "coverage.dat").resolve()))
+    count = validate(records, context(), read(PROFILE))
+    output = Path(output).resolve()
+    output.mkdir(parents=True, exist_ok=False)
+    subprocess.run([sys.executable, "-m", "coverage", "combine", "--keep", *coverage], check=True,
+                   env=dict(os.environ, COVERAGE_FILE=str(output / ".coverage")))
+    totals = {}
+    for record in records:
+        for name, seconds in record["durations"].items():
+            totals[name] = totals.get(name, 0) + seconds
+    write(output / "timings.json", {"default_seconds": 1, "seconds": totals})
+    write(output / "summary.json", {
+        "context": context(), "tests": count,
+        "shards": [{"shard": r["shard"], "attempt": r["attempt"], "tests": len(r["selected"]),
+                    "seconds": r["seconds"], "python": r["python"]} for r in records],
+    })
+    print(f"Verified all {count} unit cases exactly once across four shards; coverage combined.")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--history", required=True)
+    parser.add_argument("--output", required=True)
+    parser.add_argument("--prefix", default="unit-shard")
+    args = parser.parse_args()
+    aggregate(args.history, args.output, args.prefix)
diff --git a/azext_iot/tests/digitaltwins/test_dt_resource_unit.py b/azext_iot/tests/digitaltwins/test_dt_resource_unit.py
index 4ef152dd1..d032a07be 100644
--- a/azext_iot/tests/digitaltwins/test_dt_resource_unit.py
+++ b/azext_iot/tests/digitaltwins/test_dt_resource_unit.py
@@ -9,7 +9,6 @@
 import responses
 import json
 import azext_iot.digitaltwins.providers.resource
-from time import sleep
 from azext_iot.digitaltwins import commands_resource as subject
 from azext_iot.tests.digitaltwins.dt_helpers import generate_generic_id
 from msrestazure.azure_exceptions import CloudError
@@ -70,6 +69,7 @@ def service_client_with_retry(self, mocked_response, start_twin_response):
             body=provisioning,
             status=201,
             headers={
+                "Retry-After": "0",
                 "Azure-AsyncOperation":
                     "https://management.azure.com/subscriptions/xxx/providers/Microsoft.DigitalTwins/"
                     "locations/xxx/operationResults/operationkey"
@@ -92,14 +92,15 @@ def service_client_with_retry(self, mocked_response, start_twin_response):
 
     def test_create_instance_with_retry(self, fixture_cmd, mocker, service_client_with_retry):
         mocker.patch.object(azext_iot.digitaltwins.providers.generic, "ADT_CREATE_RETRY_AFTER", 0.0001)
-        subject.create_instance(
+        result = subject.create_instance(
             cmd=fixture_cmd,
             name=name,
             resource_group_name=resource_group_name,
             location=location
         )
-        while len(service_client_with_retry.calls) == 1:
-            sleep(10)
+        result.result(timeout=10)
+        assert result.done()
+        assert result.status() == "Succeeded"
         check_request = service_client_with_retry.calls[1].request
         assert "operationkey" in check_request.url
 
@@ -120,6 +121,7 @@ def service_client_with_failed_retry(self, mocked_response, start_twin_response)
             body=provisioning,
             status=201,
             headers={
+                "Retry-After": "0",
                 "Azure-AsyncOperation":
                     "https://management.azure.com/subscriptions/xxx/providers/Microsoft.DigitalTwins/"
                     "locations/xxx/operationResults/operationkey"
@@ -148,18 +150,13 @@ def test_create_instance_with_failed_retry(self, fixture_cmd, mocker, service_cl
             resource_group_name=resource_group_name,
             location=location
         )
-        while len(service_client_with_failed_retry.calls) == 1:
-            sleep(10)
+        with pytest.raises(CloudError):
+            result.result(timeout=10)
+        assert result.done()
         check_request = service_client_with_failed_retry.calls[1].request
         assert "operationkey" in check_request.url
 
         # The LRO poller calls once more for some reason
         assert len(service_client_with_failed_retry.calls) >= 2
         assert service_client_with_failed_retry.calls[1].response.content.decode("utf-8") == failed
-        # Sleep to give time for the poller
-        sleep(1)
-
-        # The poller.result will have the error
         assert result.status() == "Failed"
-        with pytest.raises(CloudError):
-            result.result()
diff --git a/azext_iot/tests/test_ado_dps_wiring_unit.py b/azext_iot/tests/test_ado_dps_wiring_unit.py
index cf1fdfabf..3adf594e8 100644
--- a/azext_iot/tests/test_ado_dps_wiring_unit.py
+++ b/azext_iot/tests/test_ado_dps_wiring_unit.py
@@ -49,9 +49,8 @@ def test_ado_hub_suite_has_a_nonempty_constrained_unset_default():
     assert set(conditions) == {"'sentinel'"}
 
 
-@pytest.mark.parametrize("caller", ["merge.yml", "templates/trigger-tests.yml"])
-def test_ado_callers_keep_explicit_owned_suites_and_unset_non_hub_defaults(caller):
-    pipeline = yaml.safe_load((ROOT / ".azure-devops" / caller).read_text(encoding="utf-8"))
+def test_ado_integration_callers_keep_explicit_owned_suites_and_unset_non_hub_defaults():
+    pipeline = yaml.safe_load((ROOT / ".azure-devops/templates/trigger-tests.yml").read_text(encoding="utf-8"))
     calls = [
         (job["job"], step["parameters"])
         for job in pipeline["jobs"] for step in job.get("steps", [])
@@ -63,13 +62,6 @@ def test_ado_callers_keep_explicit_owned_suites_and_unset_non_hub_defaults(calle
             assert parameters["hubSuite"] == job
         else:
             assert "hubSuite" not in parameters
-        if caller == "merge.yml":
-            assert parameters["runUnitTests"] is True
-            assert parameters["runIntTests"] is False
-    if caller == "merge.yml":
-        assert {job for job, _ in calls} == {
-            "run_unit_tests_ubuntu", "run_unit_tests_macOs", "run_unit_tests_windows",
-        }
 
 
 def _script():
diff --git a/azext_iot/tests/test_unit_shards_unit.py b/azext_iot/tests/test_unit_shards_unit.py
new file mode 100644
index 000000000..b4f06a0a9
--- /dev/null
+++ b/azext_iot/tests/test_unit_shards_unit.py
@@ -0,0 +1,406 @@
+# coding=utf-8
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+"""Unit shard collection, artifact isolation, aggregation and pipeline wiring contracts."""
+
+from concurrent.futures import ThreadPoolExecutor
+from copy import deepcopy
+from fnmatch import fnmatchcase
+from itertools import product
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+
+import pytest
+import yaml
+
+from azext_iot.tests import _unit_shards as shards
+
+ROOT = Path(__file__).resolve().parents[2]
+PROFILE = {"seconds": {}, "default_seconds": 1}
+CONTEXT = {"build": "offline-unit-shards", "commit": "a" * 40}
+
+
+def records():
+    inventory = [f"azext_iot/tests/test_{i}_unit.py::test_case::0" for i in range(4)]
+    return [{
+        "schema": 1, "context": CONTEXT, "python": "3.13.0", "shard": i + 1, "attempt": 1,
+        "profile": shards.digest(PROFILE), "inventory": inventory, "selected": [node],
+        "finished": True, "exitstatus": 0, "seconds": 1,
+        "reports": {node: {"setup": ["passed"], "call": ["passed"], "teardown": ["passed"]}},
+        "durations": {node.split("::")[0]: 1},
+    } for i, node in enumerate(inventory)]
+
+
+def test_balancing_is_deterministic_and_includes_unknown_new_files():
+    files = [f"file-{i}" for i in range(8)]
+    profile = {"seconds": dict(zip(files, [90, 80, 70, 60, 40, 30, 20, 10])), "default_seconds": 1}
+    plan = shards.partition(files, profile)
+    assert plan == shards.partition(reversed(files), profile)
+    assert [sum(profile["seconds"][name] for name in part) for part in plan] == [100] * 4
+    updated = shards.partition(files + ["new-file"], profile)
+    assert sorted(name for part in updated for name in part) == sorted(files + ["new-file"])
+
+
+@pytest.mark.parametrize("weight", [0, -1, float("nan"), float("inf"), "1"])
+def test_bad_weights_cannot_silently_change_selection(weight):
+    with pytest.raises(ValueError):
+        shards.partition(["a", "b", "c", "d"], {"seconds": {"a": weight}, "default_seconds": 1})
+
+
+@pytest.mark.parametrize("defect", [
+    "missing-shard", "duplicate-shard", "context", "python", "profile", "inventory", "selection",
+    "unfinished", "exitstatus", "missing-report", "failed-call", "missing-teardown", "duplicate-stage",
+])
+def test_aggregate_rejects_incomplete_or_inconsistent_unit_execution(defect):
+    values = deepcopy(records())
+    record = values[0]
+    node = record["selected"][0]
+    if defect == "missing-shard":
+        values.pop()
+    elif defect == "duplicate-shard":
+        values[-1]["shard"] = 1
+    elif defect in ("context", "python", "profile"):
+        record[defect] = "changed"
+    elif defect == "inventory":
+        record["inventory"] = record["inventory"][:-1]
+    elif defect == "selection":
+        record["selected"] = values[1]["selected"]
+    elif defect == "unfinished":
+        record["finished"] = False
+    elif defect == "exitstatus":
+        record["exitstatus"] = 1
+    elif defect == "missing-report":
+        record["reports"] = {}
+    elif defect == "failed-call":
+        record["reports"][node]["call"] = ["failed"]
+    elif defect == "missing-teardown":
+        del record["reports"][node]["teardown"]
+    else:
+        record["reports"][node]["setup"].append("passed")
+    with pytest.raises(ValueError):
+        shards.validate(values, CONTEXT, PROFILE)
+
+
+def test_unit_gate_accepts_different_patch_releases_within_the_requested_python_series():
+    values = records()
+    for record, version in zip(values, ["3.13.15", "3.13.15", "3.13.16", "3.13.15"]):
+        record["python"] = version
+    assert shards.validate(values, CONTEXT, PROFILE) == 4
+
+
+@pytest.mark.parametrize("version", ["3.12.16", "3.14.0", "4.13.0", "3.13", "3.13.bad", "3.13.16rc1", "", None])
+def test_unit_gate_rejects_other_python_series_and_malformed_versions(version):
+    values = records()
+    values[2]["python"] = version
+    with pytest.raises(ValueError, match="Python"):
+        shards.validate(values, CONTEXT, PROFILE)
+
+
+def test_unit_gate_preserves_standard_pytest_skip_semantics():
+    values = records()
+    node = values[0]["selected"][0]
+    values[0]["reports"][node] = {"setup": ["skipped"], "teardown": ["passed"]}
+    assert shards.validate(values, CONTEXT, PROFILE) == 4
+
+
+def make_artifacts(root, prefix="unit-shard"):
+    for record in records():
+        folder = root / f"{prefix}-{record['shard']}-1"
+        folder.mkdir()
+        for name in shards.ARTIFACTS:
+            (folder / name).write_bytes(b"offline artifact")
+        record["artifacts"] = {name: shards.file_digest(folder / name) for name in shards.ARTIFACTS}
+        shards.write(folder / "receipt.json", record)
+
+
+@pytest.mark.parametrize("prefix", ["unit-shard", "tox-unit-windows-2025-py3.13"])
+@pytest.mark.parametrize("defect", ["missing", "changed", "empty-newer-attempt", "wrong-attempt", "extra-shard"])
+def test_artifact_gate_never_falls_back_to_older_or_unverified_results(tmp_path, defect, prefix):
+    make_artifacts(tmp_path, prefix)
+    folder = tmp_path / f"{prefix}-1-1"
+    if defect == "missing":
+        (folder / "coverage.dat").unlink()
+    elif defect == "changed":
+        (folder / "junit.xml").write_text("changed")
+    elif defect == "empty-newer-attempt":
+        (tmp_path / f"{prefix}-1-2").mkdir()
+    elif defect == "wrong-attempt":
+        folder.rename(tmp_path / f"{prefix}-1-2")
+    else:
+        (tmp_path / f"{prefix}-5-1").mkdir()
+    with pytest.raises((ValueError, FileNotFoundError)):
+        shards.aggregate(tmp_path, tmp_path / "combined", prefix)
+    assert not (tmp_path / "combined").exists()
+
+
+@pytest.mark.parametrize("prefix", ["unit-shard", "tox-unit-windows-2025-py3.13"])
+def test_latest_successful_native_unit_attempt_is_aggregated(tmp_path, monkeypatch, mocker, prefix):
+    make_artifacts(tmp_path, prefix)
+    make_artifacts(tmp_path, "tox-unit-macos-15-intel-py3.12")
+    folder = tmp_path / f"{prefix}-1-1"
+    folder.rename(tmp_path / f"{prefix}-1-2")
+    folder = tmp_path / f"{prefix}-1-2"
+    record = shards.read(folder / "receipt.json")
+    record["attempt"] = 2
+    record["python"] = "3.13.16"
+    shards.write(folder / "receipt.json", record)
+    profile = tmp_path / "profile.json"
+    shards.write(profile, PROFILE)
+    monkeypatch.setattr(shards, "PROFILE", profile)
+    monkeypatch.setenv("UNIT_RUN_ID", CONTEXT["build"])
+    monkeypatch.setenv("UNIT_COMMIT", CONTEXT["commit"])
+    run = mocker.patch.object(shards.subprocess, "run")
+    shards.aggregate(tmp_path, tmp_path / "combined", prefix)
+    summary = shards.read(tmp_path / "combined/summary.json")
+    assert summary["tests"] == 4
+    assert [record["python"] for record in summary["shards"]] == ["3.13.16", "3.13.0", "3.13.0", "3.13.0"]
+    assert len([arg for arg in run.call_args.args[0] if arg.endswith("coverage.dat")]) == 4
+    assert f"{prefix}-1-2" in run.call_args.args[0][5]
+
+
+@pytest.mark.parametrize("prefix", ["unit-shard", "tox-unit-windows-2025-py3.13"])
+def test_real_four_process_collection_and_coverage_with_random_parameters(tmp_path, monkeypatch, prefix):
+    directory = tmp_path / "azext_iot/tests"
+    directory.mkdir(parents=True)
+    (tmp_path / "pytest.ini").write_text("[pytest]\njunit_family=xunit1\n", encoding="utf-8")
+    (tmp_path / "sample.py").write_text("def value():\n    return 1\n", encoding="utf-8")
+    for index in range(4):
+        (directory / f"test_{index}_unit.py").write_text(
+            "import uuid, pytest, sample\n"
+            "@pytest.mark.parametrize('value', [str(uuid.uuid4()), str(uuid.uuid4())])\n"
+            "def test_example(value):\n"
+            "    assert value and sample.value() == 1\n",
+            encoding="utf-8",
+        )
+
+    def execute(index):
+        output = tmp_path / f"{prefix}-{index}-1"
+        env = dict(os.environ, UNIT_RUN_ID=CONTEXT["build"], UNIT_COMMIT=CONTEXT["commit"],
+                   UNIT_ATTEMPT="1", COVERAGE_FILE=str(output / "coverage.dat"), PYTEST_ADDOPTS="",
+                   PYTHONPATH=os.pathsep.join(dict.fromkeys([str(tmp_path), str(ROOT), *map(os.path.abspath, sys.path)])))
+        result = subprocess.run(
+            [sys.executable, "-m", "pytest", str(directory), "-c", str(tmp_path / "pytest.ini"), "-k", "_unit.py",
+             "-p", "azext_iot.tests._unit_shard_plugin", "--unit-shard", str(index), "--unit-shard-output", str(output),
+             "--cov=sample", "--cov-report=", "--junitxml", str(output / "junit.xml"), "-q"],
+            cwd=tmp_path, env=env, capture_output=True, text=True, timeout=90, check=False,
+        )
+        assert result.returncode == 0, result.stdout + result.stderr
+        return shards.read(output / "receipt.json")
+
+    with ThreadPoolExecutor(max_workers=4) as pool:
+        values = list(pool.map(execute, range(1, 5)))
+    assert shards.validate(values, CONTEXT, shards.read(shards.PROFILE)) == 8
+    assert all(len(record["selected"]) == 2 for record in values)
+    monkeypatch.setenv("UNIT_RUN_ID", CONTEXT["build"])
+    monkeypatch.setenv("UNIT_COMMIT", CONTEXT["commit"])
+    shards.aggregate(tmp_path, tmp_path / "combined", prefix)
+    assert (tmp_path / "combined/.coverage").is_file()
+    assert len(shards.read(tmp_path / "combined/timings.json")["seconds"]) == 4
+
+
+def test_github_prechecks_keep_lint_independent_and_gate_every_shard():
+    jobs = yaml.safe_load((ROOT / ".github/workflows/int_test.yml").read_text(encoding="utf-8"))["jobs"]
+    lint, job, gate = (jobs[name] for name in ("lint", "unit-shards", "unit-test"))
+    assert "needs" not in lint and "needs" not in job
+    assert job["strategy"] == {"fail-fast": False, "max-parallel": 4, "matrix": {"shard": [1, 2, 3, 4]}}
+    assert gate["needs"] == ["lint", "unit-shards"]
+    for precheck in (lint, job, gate):
+        assert "if" not in precheck and "continue-on-error" not in precheck
+        assert precheck["permissions"] == {"contents": "read"}
+        assert not any(step.get("continue-on-error") for step in precheck["steps"])
+    run = next(step for step in job["steps"] if "run" in step)
+    assert "-e python-azcur-unit" in run["run"]
+    assert "--unit-shard" in run["run"] and "--unit-shard-output" in run["run"]
+    assert run["env"]["COVERAGE_FILE"].endswith("/coverage.dat")
+    assert run["env"]["UNIT_ATTEMPT"] == "${{ github.run_attempt }}"
+    assert "if" not in run
+    upload = next(step for step in job["steps"] if step.get("uses", "").startswith("actions/upload-artifact@"))
+    assert upload["if"] == "${{ always() }}"
+    assert upload["with"]["name"] == "unit-shard-${{ matrix.shard }}-${{ github.run_attempt }}"
+    assert upload["with"]["if-no-files-found"] == "error" and not upload["with"].get("overwrite")
+    download = next(step for step in gate["steps"] if step.get("uses", "").startswith("actions/download-artifact@"))
+    assert download["with"] == {"pattern": "unit-shard-*", "path": "unit-history"}
+    combine = next(step for step in gate["steps"] if "run" in step)
+    assert "_unit_shards.py --history unit-history --output unit-coverage" in combine["run"]
+    assert combine["env"] == {name: run["env"][name] for name in ("UNIT_RUN_ID", "UNIT_COMMIT")}
+    coverage = next(step for step in gate["steps"] if step.get("with", {}).get("name") == "coverage-unit")
+    assert coverage["with"]["path"] == "unit-coverage/.coverage"
+    assert coverage["with"]["include-hidden-files"] is True
+    assert coverage["with"]["if-no-files-found"] == "error" and "if" not in coverage
+    assert jobs["int-test"]["needs"] == ["setup", "unit-test"]
+    assert "needs.unit-test.result == 'success'" in jobs["int-test"]["if"]
+
+
+@pytest.mark.parametrize("branch,push_enabled", [
+    ("dev", True),
+    ("preview", True),
+    ("1.1.0-preview", True),
+    ("release/1.0.0-preview", True),
+    ("release/1.1.0-preview", True),
+    ("release/future-preview", True),
+    ("users/hangyiwang/unit-test-parallelism", False),
+    ("users/hangyiwang/ado147-integration-parity-retries", False),
+    ("dependabot/github_actions/update", False),
+    ("fix/example", False),
+])
+def test_pr_ci_does_not_duplicate_feature_branch_pushes(branch, push_enabled):
+    workflow = yaml.safe_load((ROOT / ".github/workflows/ci_workflow.yml").read_text(encoding="utf-8"))
+    # PyYAML's YAML 1.1 parser also recognizes unquoted "on" as True.
+    events = workflow.get("on", workflow.get(True))
+    assert events == {
+        "pull_request": None,
+        "push": {"branches": ["dev", "preview", "1.1.0-preview", "release/**"], "tags": ["**"]},
+        "workflow_dispatch": None,
+    }
+    assert any(fnmatchcase(branch, pattern) for pattern in events["push"]["branches"]) == push_enabled
+
+
+def test_ado_merge_retains_non_unit_checks_without_repeating_github_units():
+    workflow = yaml.safe_load((ROOT / ".azure-devops/merge.yml").read_text(encoding="utf-8"))
+    jobs = {job["job"]: job for job in workflow["jobs"]}
+    assert set(jobs) == {
+        "build_and_publish_azure_iot_cli_ext", "build_and_publish_azure_cli_test_sdk",
+        "run_style_check", "run_azdev_linter_on_command_table", "CredScan",
+    }
+    for job in jobs.values():
+        assert not job.get("continueOnError") and "condition" not in job
+        assert not any(
+            step.get("template", "").endswith("run-tests-parallel.yml")
+            or step.get("parameters", {}).get("runUnitTests")
+            for step in job["steps"]
+        )
+        assert set(job.get("dependsOn", [])) <= set(jobs)
+    scan, analyze = jobs["CredScan"]["steps"]
+    assert scan["task"] == "CredScan@3"
+    assert analyze["task"] == "PostAnalysis@1" and analyze["inputs"]["CredScan"] is True
+    build = yaml.safe_load(
+        (ROOT / ".azure-devops/templates/build-publish-azure-iot-cli-extension.yml").read_text(encoding="utf-8")
+    )
+    assert any("ManifestGeneratorTask@" in step.get("task", "") for step in build["steps"])
+    assert jobs["build_and_publish_azure_iot_cli_ext"]["steps"][-1]["template"] == (
+        "templates/build-publish-azure-iot-cli-extension.yml"
+    )
+    assert jobs["build_and_publish_azure_cli_test_sdk"]["steps"][-1]["template"] == (
+        "templates/build-publish-azure-cli-test-sdk.yml"
+    )
+    assert jobs["run_azdev_linter_on_command_table"]["steps"][-1]["template"] == "templates/evaluate-command-table.yml"
+
+
+def test_only_ci_pull_requests_request_the_reduced_unit_matrix():
+    workflow = yaml.safe_load((ROOT / ".github/workflows/tox.yml").read_text(encoding="utf-8"))
+    events = workflow.get("on", workflow.get(True))
+    option = events["workflow_call"]["inputs"]["pr-matrix"]
+    assert option["type"] == "boolean" and option["default"] is False and option["required"] is False
+    assert "pr-matrix" not in events["workflow_dispatch"]["inputs"]
+    callers = {
+        (path.name, name): job
+        for path in (ROOT / ".github/workflows").glob("*.yml")
+        for name, job in yaml.safe_load(path.read_text(encoding="utf-8")).get("jobs", {}).items()
+        if job.get("uses") == "./.github/workflows/tox.yml"
+    }
+    assert ("release_workflow.yml", "unit-test") in callers
+    for key, job in callers.items():
+        if key == ("ci_workflow.yml", "test"):
+            assert job["with"]["pr-matrix"] == "${{ github.event_name == 'pull_request' }}"
+        else:
+            assert job.get("with", {}).get("pr-matrix", False) is False
+
+
+@pytest.mark.parametrize("pr_matrix", [False, True])
+def test_unit_and_gate_matrices_cover_exact_combinations(pr_matrix):
+    jobs = yaml.safe_load((ROOT / ".github/workflows/tox.yml").read_text(encoding="utf-8"))["jobs"]
+    expected_pairs = {
+        ("ubuntu-24.04", "3.10"), ("ubuntu-24.04", "3.13"),
+        ("windows-2025", "3.13"), ("macos-15-intel", "3.13"),
+    } if pr_matrix else set(product(
+        ("ubuntu-24.04", "windows-2025", "macos-15-intel"), ("3.10", "3.11", "3.12", "3.13")
+    ))
+    for name in ("tox", "unit-gate"):
+        matrix = jobs[name]["strategy"]["matrix"]
+        expression = re.fullmatch(
+            r"\$\{\{ fromJSON\(inputs\.pr-matrix && '([^']+)' \|\| '([^']+)'\) \}\}",
+            matrix["exclude"],
+        )
+        assert expression is not None
+        exclusions = json.loads(expression.group(1 if pr_matrix else 2))
+        axes = {key: values for key, values in matrix.items() if key != "exclude"}
+        expanded = [dict(zip(axes, values)) for values in product(*axes.values())]
+        selected = [
+            item for item in expanded
+            if not any(all(item[key] == value for key, value in exclusion.items()) for exclusion in exclusions)
+        ]
+        actual = {(item["os"], item["py"], item.get("shard")) for item in selected}
+        expected_shards = [1, 2, 3, 4] if name == "tox" else [None]
+        assert actual == {
+            (os_name, python, shard)
+            for os_name, python in expected_pairs for shard in expected_shards
+        }
+        assert len(selected) == len(actual) == (4 if pr_matrix else 12) * len(expected_shards)
+
+
+def test_ci_shards_selected_os_python_combinations_without_mixing_artifacts():
+    caller = yaml.safe_load((ROOT / ".github/workflows/ci_workflow.yml").read_text(encoding="utf-8"))["jobs"]["test"]
+    assert caller["uses"] == "./.github/workflows/tox.yml"
+    assert caller.get("name", "test") == "test"
+    assert caller["with"]["pr-matrix"] == "${{ github.event_name == 'pull_request' }}"
+    assert caller.get("with", {}).get("continue-on-error", False) is False
+    jobs = yaml.safe_load((ROOT / ".github/workflows/tox.yml").read_text(encoding="utf-8"))["jobs"]
+    unit, gate = jobs["tox"], jobs["unit-gate"]
+    assert gate["name"] == "Unit test ${{ matrix.py }} - ${{ matrix.os }}"
+    assert unit["name"] == gate["name"] + " (shard ${{ matrix.shard }}/4)"
+    matrix = {
+        "os": ["ubuntu-24.04", "windows-2025", "macos-15-intel"],
+        "py": ["3.13", "3.12", "3.11", "3.10"],
+        "exclude": unit["strategy"]["matrix"]["exclude"],
+    }
+    assert unit["strategy"] == {"fail-fast": False, "matrix": dict(matrix, shard=[1, 2, 3, 4])}
+    assert gate["strategy"] == {"fail-fast": False, "matrix": matrix}
+    assert gate["needs"] == "tox" and gate["if"] == "${{ always() }}"
+    guard = gate["steps"][0]
+    assert guard["name"] == "Require successful unit shards and lint"
+    assert guard["if"] == "${{ needs.tox.result != 'success' }}"
+    assert guard["run"].strip().endswith("exit 1") and "::error::" in guard["run"]
+    assert "continue-on-error" not in guard
+    for job in (unit, gate):
+        assert job["runs-on"] == "${{ matrix.os }}"
+        assert job["continue-on-error"] == "${{ inputs.continue-on-error }}"
+    run = next(step for step in unit["steps"] if step.get("name") == "Run test suite")
+    assert "-e python-azcur-unit --skip-pkg-install --" in run["run"]
+    assert "--unit-shard ${{ matrix.shard }}" in run["run"]
+    assert run["env"]["UNIT_ATTEMPT"] == "${{ github.run_attempt }}"
+    assert ":tox:${{ matrix.os }}:py${{ matrix.py }}" in run["env"]["UNIT_RUN_ID"]
+    lint = next(step for step in unit["steps"] if step.get("name") == "Run lint once per OS and Python")
+    assert lint["if"] == "${{ matrix.shard == 1 }}" and "-e lint" in lint["run"]
+    upload = next(step for step in unit["steps"] if step.get("name") == "Upload unit shard evidence")
+    prefix = "tox-unit-${{ matrix.os }}-py${{ matrix.py }}"
+    assert upload["with"]["name"] == prefix + "-${{ matrix.shard }}-${{ github.run_attempt }}"
+    assert upload["if"] == "${{ always() }}" and upload["with"]["if-no-files-found"] == "error"
+    assert not upload["with"].get("overwrite")
+    download = next(step for step in gate["steps"] if step.get("uses", "").startswith("actions/download-artifact@"))
+    assert download["with"] == {"pattern": prefix + "-*", "path": "unit-history"}
+    combine = next(step for step in gate["steps"] if "_unit_shards.py" in step.get("run", ""))
+    assert "--prefix " + prefix in combine["run"]
+    assert combine["env"] == {name: run["env"][name] for name in ("UNIT_RUN_ID", "UNIT_COMMIT")}
+    report = next(step for step in gate["steps"] if step.get("name") == "Generate coverage reports")
+    assert report["env"]["COVERAGE_FILE"].endswith("/unit-coverage/.coverage")
+    assert all(f"coverage {kind}" in report["run"] for kind in ("report", "html", "json"))
+    coverage = next(step for step in gate["steps"] if step.get("with", {}).get("name") == "code-coverage")
+    assert coverage["if"] == "${{ matrix.os == 'ubuntu-24.04' && matrix.py == '3.13' }}"
+    assert coverage["with"]["path"] == "htmlcov/"
+
+
+def test_tox_passes_ci_neutral_shard_identity_without_changing_default_unit_selection():
+    from configparser import ConfigParser
+    config = ConfigParser(interpolation=None)
+    config.read(ROOT / "tox.ini")
+    unit = config["testenv:py{thon,38,39,310,311,312,313}-az{min,cur,dev}-unit"]
+    assert {"UNIT_RUN_ID", "UNIT_COMMIT", "UNIT_ATTEMPT", "COVERAGE_FILE"} <= set(unit["passenv"].split())
+    assert 'python -m pytest -k "_unit.py" ./azext_iot/tests' in unit["commands"]
+    assert "{posargs}" in unit["commands"]
diff --git a/azext_iot/tests/test_workflow_parallelism_unit.py b/azext_iot/tests/test_workflow_parallelism_unit.py
index 23909a446..d31cfddbd 100644
--- a/azext_iot/tests/test_workflow_parallelism_unit.py
+++ b/azext_iot/tests/test_workflow_parallelism_unit.py
@@ -101,7 +101,7 @@ def test_matrix_rejects_untrusted_selector_or_canary_region_mismatch(tmp_path, m
 
 def test_flat_matrix_makes_all_combinations_concurrently_eligible_without_wrappers_or_locks():
     public = _workflows()[0]
-    assert set(public["jobs"]) == {"setup", "unit-test", "int-test", "int-test-gate", "combine-coverage"}
+    assert set(public["jobs"]) == {"setup", "lint", "unit-shards", "unit-test", "int-test", "int-test-gate", "combine-coverage"}
     job = public["jobs"]["int-test"]
     assert job["needs"] == ["setup", "unit-test"]
     assert job["if"] == (
@@ -115,9 +115,10 @@ def test_flat_matrix_makes_all_combinations_concurrently_eligible_without_wrappe
     }
     assert public["jobs"]["setup"]["outputs"] == {"matrix": "${{ steps.matrix.outputs.matrix }}"}
     assert "concurrency" not in public
-    for value in public["jobs"].values():
+    for name, value in public["jobs"].items():
         assert "concurrency" not in value
-        assert "max-parallel" not in value.get("strategy", {})
+        if name != "unit-shards":
+            assert "max-parallel" not in value.get("strategy", {})
         assert not value.get("continue-on-error", False)
     for name in ("int_test_bundle.yml", "int_test_cohort.yml"):
         assert not (ROOT / ".github/workflows" / name).exists()
diff --git a/azext_iot/tests/unit_test_durations.json b/azext_iot/tests/unit_test_durations.json
new file mode 100644
index 000000000..79ac3ed56
--- /dev/null
+++ b/azext_iot/tests/unit_test_durations.json
@@ -0,0 +1,58 @@
+{
+  "source": "2026-10-05 complete local unit JUnit reports; files below two seconds use the default",
+  "default_seconds": 1,
+  "seconds": {
+    "azext_iot/tests/adr/test_adr_base_unit.py": 2.96,
+    "azext_iot/tests/adr/test_adr_certificate_action_tracker_unit.py": 7.84,
+    "azext_iot/tests/adr/test_adr_certificate_activation_unit.py": 4.11,
+    "azext_iot/tests/adr/test_adr_certificate_authority_unit.py": 2.63,
+    "azext_iot/tests/adr/test_adr_certificate_policy_unit.py": 3.39,
+    "azext_iot/tests/adr/test_adr_job_run_scenarios_unit.py": 8.5,
+    "azext_iot/tests/adr/test_adr_job_run_unit.py": 2.69,
+    "azext_iot/tests/adr/test_adr_link_delete_unit.py": 6.61,
+    "azext_iot/tests/adr/test_adr_link_propagation_unit.py": 2.7,
+    "azext_iot/tests/adr/test_adr_link_sequential_unit.py": 5.18,
+    "azext_iot/tests/adr/test_adr_link_unit.py": 6.34,
+    "azext_iot/tests/adr/test_adr_namespace_unit.py": 4.41,
+    "azext_iot/tests/adr/test_adr_registry_device_contract_unit.py": 13.59,
+    "azext_iot/tests/adr/test_adr_registry_device_unit.py": 3.31,
+    "azext_iot/tests/adr/test_adr_sdk_unit.py": 8.41,
+    "azext_iot/tests/adr/test_adr_surface_retirement_unit.py": 6.82,
+    "azext_iot/tests/adr/test_adr_validation_scenarios_unit.py": 9.1,
+    "azext_iot/tests/digitaltwins/test_dt_model_unit.py": 2.58,
+    "azext_iot/tests/digitaltwins/test_dt_twin_unit.py": 5.19,
+    "azext_iot/tests/dps/core/test_dps_phase_lifetime_unit.py": 5.91,
+    "azext_iot/tests/dps/core/test_dps_phase_runtime_unit.py": 51.31,
+    "azext_iot/tests/dps/device_registration/test_csr_issuance_fixture_unit.py": 3.05,
+    "azext_iot/tests/dps/device_registration/test_csr_registry_cleanup_unit.py": 13.82,
+    "azext_iot/tests/dps/device_registration/test_registry_assertions_unit.py": 2.03,
+    "azext_iot/tests/dps/enrollment/test_iot_dps_enrollment_unit.py": 30.88,
+    "azext_iot/tests/dps/enrollment_group/test_iot_dps_enrollment_group_unit.py": 35,
+    "azext_iot/tests/dps/test_hub_dps_consolidation_unit.py": 26.29,
+    "azext_iot/tests/dps/test_registration_recovery_unit.py": 24.55,
+    "azext_iot/tests/iothub/configurations/test_iot_config_unit.py": 13.61,
+    "azext_iot/tests/iothub/core/test_iot_ext_unit.py": 70.95,
+    "azext_iot/tests/iothub/devices/test_iot_device_unit.py": 35.01,
+    "azext_iot/tests/iothub/devices/test_iot_edge_devices_unit.py": 100.3,
+    "azext_iot/tests/iothub/jobs/test_iothub_jobs_unit.py": 18.37,
+    "azext_iot/tests/iothub/state/test_hub_state_integration_unit.py": 4.32,
+    "azext_iot/tests/iothub/test_dynamic_hub_cleanup_unit.py": 34.19,
+    "azext_iot/tests/iothub/test_integration_fixtures_unit.py": 2.5,
+    "azext_iot/tests/iothub/test_preview_auth_switch_unit.py": 4.79,
+    "azext_iot/tests/iothub/test_preview_authorization_wire_unit.py": 3.48,
+    "azext_iot/tests/iothub/test_sas_phase_unit.py": 4.03,
+    "azext_iot/tests/test_ado_dps_wiring_unit.py": 2.7,
+    "azext_iot/tests/test_command_loader_unit.py": 35.02,
+    "azext_iot/tests/test_dps_phase_runner_unit.py": 9.19,
+    "azext_iot/tests/test_focused_live_runner_unit.py": 76.21,
+    "azext_iot/tests/test_hub_observer_polling_unit.py": 7.19,
+    "azext_iot/tests/test_hub_ownership_transport_unit.py": 9.46,
+    "azext_iot/tests/test_hub_phase_runner_unit.py": 143.11,
+    "azext_iot/tests/test_hub_suite_unit.py": 10.37,
+    "azext_iot/tests/test_integration_reporting_process_unit.py": 2.19,
+    "azext_iot/tests/test_integration_results_unit.py": 3.14,
+    "azext_iot/tests/test_refresh_ci_auth_unit.py": 42.05,
+    "azext_iot/tests/test_workflow_parallelism_unit.py": 10.71,
+    "azext_iot/tests/test_workflow_results_unit.py": 17.79
+  }
+}
diff --git a/docs/tox-testing.md b/docs/tox-testing.md
index f1e0434c9..ac125862c 100644
--- a/docs/tox-testing.md
+++ b/docs/tox-testing.md
@@ -83,8 +83,13 @@ In order to list all recognized environments, you can type `tox -av`, which will
 
 ## Integration workflow topology
 
-`int_test.yml` runs a direct service × Python × region matrix after setup and unit
-tests succeed. Every selected combination, including ADU, is independently eligible
+`int_test.yml` runs lint and four duration-balanced serial unit shards in parallel.
+Its unit gate verifies complete, nonduplicated execution and combines coverage
+before any integration jobs are eligible. See [parallel CI unit tests](../CONTRIBUTING.md#parallel-ci-unit-tests)
+for shard artifacts, reruns and timing-profile maintenance.
+
+The direct service × Python × region matrix starts after setup and the unit gate
+succeed. Every selected combination, including ADU, is independently eligible
 to run, with `fail-fast: false`, no parallelism cap and no workflow/job concurrency
 lock. Job names and result/coverage artifacts identify the service, Python and region.
 The result gate checks the complete selected matrix independently of coverage reporting.
diff --git a/tox.ini b/tox.ini
index 411ed0983..094e15a01 100644
--- a/tox.ini
+++ b/tox.ini
@@ -55,6 +55,10 @@ setenv =
     PYTHONPATH={envsitepackagesdir}/azure-cli-extensions/azure-iot
 passenv =
     azext_*
+    UNIT_RUN_ID
+    UNIT_COMMIT
+    UNIT_ATTEMPT
+    COVERAGE_FILE
 deps =
     # base deps
     {[base]deps}
@@ -73,7 +77,7 @@ commands =
     az -v
     # run tests
     # You can pass additional positional args to pytest using `-- {args}`
-    pytest -k "_unit.py" ./azext_iot/tests \
+    python -m pytest -k "_unit.py" ./azext_iot/tests \
     --cov=azext_iot --cov-append --cov-config .coveragerc --junitxml=junit/test-iotext-unit.xml {posargs}
 
 [testenv:DPS-phases]