diff --git a/.github/workflows/runner-utilization.yml b/.github/workflows/runner-utilization.yml
index 3cf2856f6..d126ea40b 100644
--- a/.github/workflows/runner-utilization.yml
+++ b/.github/workflows/runner-utilization.yml
@@ -56,7 +56,10 @@ jobs:
             -p 'test_runner_utilization_report.py' -v
 
       - name: Generate Utilization Report
-        timeout-minutes: 30
+        # Full-coverage fetch (created-filtered listing + 12h lookback) is
+        # ~2-3x the truncated 5000-run fetch this replaces; the old fetch
+        # took ~10 min, so give the full one comfortable headroom.
+        timeout-minutes: 55
         env:
           GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         run: |
diff --git a/scripts/ci/utils/runner_utilization_report.py b/scripts/ci/utils/runner_utilization_report.py
index 4816aa4a1..45dfe7bd1 100755
--- a/scripts/ci/utils/runner_utilization_report.py
+++ b/scripts/ci/utils/runner_utilization_report.py
@@ -84,35 +84,98 @@ def run_gh_command(args: list[str], max_retries: int = 10) -> dict:
     raise Exception(f"gh api failed after {max_retries} attempts: {last_err[:300]}")
 
 
-def get_workflow_runs(repo: str, hours: int = 24) -> list[dict]:
-    """Get workflow runs from the last N hours."""
-    since = datetime.now(timezone.utc) - timedelta(hours=hours)
+def get_workflow_runs(
+    repo: str, since: datetime, max_pages: int = 400
+) -> tuple[list[dict], bool]:
+    """Get workflow runs created after `since`.
+
+    Returns (runs, truncated). `truncated` is True when the safety cap cut
+    the listing short, i.e. the OLDEST part of the window is missing (runs
+    come back newest-first).
+
+    The `created` filter is applied server-side so pagination ends exactly
+    when the window is exhausted. The previous implementation listed ALL
+    runs newest-first and stopped at a hard 50-page cap (5000 runs); on
+    busy days the repo creates ~15-18k runs per 24h, so the cap silently
+    dropped the oldest ~2/3 of the window while the utilization denominator
+    still assumed full 24h coverage. Worse, under queue backlog job
+    execution lags run creation by hours, so the surviving newest slice
+    held mostly still-queued jobs — saturated pools (hosts busy 24h/24h)
+    reported ~4% utilization.
+
+    GitHub API gotcha: a `created`-filtered listing serves at most 1000
+    results per query (search-style cap; page 11 comes back empty even
+    though total_count is larger). We therefore walk a cursor: whenever a
+    range holds more than 1000 runs, re-query with the range's upper bound
+    moved down to the oldest created_at fetched so far, deduping the
+    boundary overlap by run id, until the range is exhausted.
+    """
+    since_str = since.strftime("%Y-%m-%dT%H:%M:%SZ")
 
     runs = []
-    page = 1
+    seen_ids = set()
+    upper_str = None  # walking upper bound (inclusive), None = now
+    pages_used = 0
+    truncated = False
     while True:
-        data = run_gh_command(
-            [
-                f"repos/{repo}/actions/runs?per_page=100&page={page}",
-            ]
+        created_q = f"{since_str}..{upper_str}" if upper_str else f">={since_str}"
+        chunk = []
+        chunk_total = None
+        page = 1
+        while True:
+            data = run_gh_command(
+                [
+                    f"repos/{repo}/actions/runs"
+                    f"?per_page=100&page={page}&created={created_q}",
+                ]
+            )
+            pages_used += 1
+            if chunk_total is None:
+                chunk_total = data.get("total_count", 0)
+            page_runs = data.get("workflow_runs", [])
+            chunk.extend(page_runs)
+            # Stop at a short page (range exhausted), the 1000-result cap
+            # (10 full pages — page 11 would be empty), or the global
+            # request budget.
+            if len(page_runs) < 100 or page >= 10 or pages_used >= max_pages:
+                break
+            page += 1
+
+        new_runs = []
+        for r in chunk:
+            rid = r.get("id")
+            if rid in seen_ids:
+                continue
+            seen_ids.add(rid)
+            new_runs.append(r)
+        runs.extend(new_runs)
+
+        if chunk_total is not None and len(chunk) >= chunk_total:
+            break  # saw everything matching this range -> reached `since`
+        if pages_used >= max_pages:
+            truncated = True
+            break
+        # More results exist below the 1000-result cap: move the upper
+        # bound down to the oldest run fetched in this chunk and re-query.
+        chunk_created = [
+            parse_time(r.get("created_at")) for r in chunk if r.get("created_at")
+        ]
+        if not new_runs or not chunk_created:
+            truncated = True  # cursor can't advance; avoid looping forever
+            break
+        new_upper = min(chunk_created).strftime("%Y-%m-%dT%H:%M:%SZ")
+        if new_upper == upper_str:
+            truncated = True
+            break
+        upper_str = new_upper
+
+    if truncated:
+        print(
+            f"WARNING: run listing truncated at {len(runs)} runs "
+            f"(request budget {max_pages} pages exhausted or cursor "
+            f"stalled). The oldest part of the window is missing."
         )
-        page_runs = data.get("workflow_runs", [])
-
-        # Filter by time
-        for run in page_runs:
-            created_at = parse_time(run.get("created_at"))
-            if created_at and created_at >= since:
-                runs.append(run)
-            elif created_at and created_at < since:
-                # Runs are ordered by created_at desc, so we can stop
-                return runs
-
-        if len(page_runs) < 100:
-            break
-        page += 1
-        if page > 50:  # Safety limit (5000 runs)
-            break
-    return runs
+    return runs, truncated
 
 
 def get_jobs_for_run(repo: str, run_id: int) -> list[dict]:
@@ -173,6 +236,45 @@ def parse_time(time_str: str) -> datetime:
     return datetime.fromisoformat(time_str.replace("Z", "+00:00"))
 
 
+def carried_over_fingerprint(job: dict):
+    """Identity of one physical job execution, for deduping re-run carryover.
+
+    When a run is re-run, the completed jobs that were NOT re-run reappear
+    in the new attempt as new job records (new id, identical
+    name/runner/timestamps), and `filter=all` returns every attempt's
+    records. Two completed records agreeing on run, name, runner, and both
+    timestamps are one execution — a real runner cannot run two identical
+    jobs at the same instant. Returns None for non-completed jobs (their
+    records are attempt-specific, and e.g. still-queued jobs lack the
+    timestamps that make the fingerprint discriminating).
+    """
+    if job.get("status") != "completed":
+        return None
+    return (
+        job.get("run_id"),
+        job.get("name"),
+        job.get("runner_name"),
+        job.get("started_at"),
+        job.get("completed_at"),
+    )
+
+
+def union_seconds(intervals: list[tuple]) -> float:
+    """Total length in seconds of the union of (start, end) intervals."""
+    busy = 0.0
+    cur_start = cur_end = None
+    for start, end in sorted(intervals):
+        if cur_end is None or start > cur_end:
+            if cur_end is not None:
+                busy += (cur_end - cur_start).total_seconds()
+            cur_start, cur_end = start, end
+        else:
+            cur_end = max(cur_end, end)
+    if cur_end is not None:
+        busy += (cur_end - cur_start).total_seconds()
+    return busy
+
+
 def classify_job(job: dict, now: datetime):
     """Derive the queue-wait and busy interval for a single job.
 
@@ -351,16 +453,23 @@ _NON_GPU_WORKFLOW_HINTS = (
     "stale",
     "dependabot",
     "codeql",
+    # Pure API-bookkeeping workflows that fire on every PR event. "PR
+    # States" alone is ~25% of all runs on a busy day; none of these ever
+    # dispatch to a self-hosted runner.
+    "pr states",
+    "slash command",
+    "cancel",  # cancel-unfinished-pr-tests, cancel-pr-workflows-on-close
+    "model inventory",
 )
 
 
 def _likely_no_gpu_jobs(workflow_name: str) -> bool:
     """Heuristic: skip per-run job-fetch for workflows that don't dispatch
-    to self-hosted GPU runners. The GH API rate limit (~5000 req/hr per
-    token) is the bottleneck on busy 24h windows where ~4000 workflow
-    runs fire — but only a fraction of those (pr-test, nightly-test,
-    pr-test-*kernel, etc.) actually run on GPU runners. Skipping the
-    docs/lint/release runs cuts the API call budget by 2-4x.
+    to self-hosted GPU runners. The GH API rate limit is the bottleneck on
+    busy 24h windows where ~18k workflow runs fire — but only a fraction
+    of those (pr-test, nightly-test, pr-test-*kernel, etc.) actually run
+    on GPU runners. Skipping the bookkeeping/docs/lint/release runs cuts
+    the API call budget roughly in half.
     """
     if not workflow_name:
         return False
@@ -368,18 +477,83 @@ def _likely_no_gpu_jobs(workflow_name: str) -> bool:
     return any(h in n for h in _NON_GPU_WORKFLOW_HINTS)
 
 
-def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None):
-    """Calculate runner utilization metrics."""
+def calculate_utilization(
+    repo: str,
+    hours: float = 24,
+    runner_filter: str = None,
+    lookback_hours: float = None,
+):
+    """Calculate runner utilization metrics.
 
-    print(f"Fetching workflow runs from last {hours} hours...")
-    all_runs = get_workflow_runs(repo, hours)
-    runs = [r for r in all_runs if not _likely_no_gpu_jobs(r.get("name", ""))]
-    skipped = len(all_runs) - len(runs)
+    `lookback_hours` extends the run *listing* (not the analysis window)
+    back before the window start. Under queue backlog, jobs execute hours
+    after their run is created, so the busy time observed inside the window
+    largely belongs to runs created before it — measured at ~33h of
+    in-window busy time from pre-window runs on one saturated 4-host pool.
+    Busy intervals are clamped to the window, so the lookback only restores
+    missing numerator; it never inflates it.
+    """
+    fetch_start = datetime.now(timezone.utc)
+    if lookback_hours is None:
+        lookback_hours = min(12.0, hours / 2)
+    window_start_precheck = fetch_start - timedelta(hours=hours)
+    since = fetch_start - timedelta(hours=hours + lookback_hours)
+
+    print(
+        f"Fetching workflow runs from last {hours}h "
+        f"(+{lookback_hours:.1f}h lookback for long-queued jobs)..."
+    )
+    all_runs, truncated = get_workflow_runs(repo, since)
+
+    runs = []
+    skipped_non_gpu = 0
+    skipped_lookback = 0
+    for r in all_runs:
+        if _likely_no_gpu_jobs(r.get("name", "")):
+            skipped_non_gpu += 1
+            continue
+        created_at = parse_time(r.get("created_at"))
+        if created_at and created_at < window_start_precheck:
+            # Lookback region: only runs that were still active at the
+            # window start can contribute in-window busy time. A completed
+            # run whose last update predates the window finished before it
+            # — skip the (expensive) per-run job fetch. In-flight runs are
+            # always kept: their updated_at can be stale while a job is
+            # still running.
+            updated_at = parse_time(r.get("updated_at"))
+            if (
+                r.get("status") == "completed"
+                and updated_at
+                and updated_at < window_start_precheck
+            ):
+                skipped_lookback += 1
+                continue
+        runs.append(r)
     print(
         f"Found {len(all_runs)} workflow runs "
-        f"({skipped} skipped as non-GPU: docs/lint/release/etc.)"
+        f"({skipped_non_gpu} skipped as non-GPU: docs/lint/release/etc.; "
+        f"{skipped_lookback} lookback runs skipped as finished pre-window)"
     )
 
+    # If the run listing was truncated (newest-first), the oldest part of
+    # the window has no data. Shrink the analysis window to what the data
+    # actually covers so busy time isn't divided by capacity-hours we never
+    # observed — that's exactly the bug that made saturated pools report
+    # single-digit utilization.
+    coverage_hours = hours
+    if truncated and all_runs:
+        oldest_created = min(
+            parse_time(r["created_at"]) for r in all_runs if r.get("created_at")
+        )
+        coverage_hours = min(
+            hours,
+            (datetime.now(timezone.utc) - oldest_created).total_seconds() / 3600,
+        )
+        print(
+            f"Shrinking analysis window: {coverage_hours:.1f}h covered of "
+            f"{hours}h requested."
+        )
+
     # Try to get online runners from API
     print("Fetching online runners...")
     runners = get_runners(repo, online_only=True)
@@ -462,11 +636,37 @@ def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None)
     # captured once so every in-flight job is measured against a single
     # reference (matches window_end below to within processing time).
     now = datetime.now(timezone.utc)
+    window_seconds = coverage_hours * 3600
+    window_end = now
+    window_start = window_end - timedelta(hours=coverage_hours)
+
     all_job_infos = []  # one entry per job (deduped across labels) for detail views
+    seen_fingerprints = set()
     for job in all_jobs:
+        # Re-run attempts carry over the completed jobs they did NOT re-run
+        # as brand-new job records: a different job id but identical
+        # name/runner/timestamps. `filter=all` returns every attempt, so on
+        # rerun-heavy days each carried-over job was double-counted —
+        # inflating busy time to ~110% on saturated hosts and padding the
+        # pass counts. Job ids differ, so dedup by execution fingerprint.
+        fp = carried_over_fingerprint(job)
+        if fp is not None:
+            if fp in seen_fingerprints:
+                continue
+            seen_fingerprints.add(fp)
         job_info = classify_job(job, now)
         if job_info is None:
             continue
+        # Lookback runs bring in jobs that finished before the window even
+        # started. They can't contribute clamped busy time and would only
+        # pollute job counts / queue stats — drop them. A job is entirely
+        # pre-window when both its busy interval and its queue wait ended
+        # before window_start.
+        latest_activity = max(
+            t for t in (job_info["end"], job_info["queue_end"]) if t is not None
+        )
+        if latest_activity < window_start:
+            continue
         all_job_infos.append(job_info)
         runner_name = job_info["runner_name"]
 
@@ -497,21 +697,22 @@ def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None)
 
     print(f"Tracking {len(all_labels)} runner labels: {sorted(all_labels)}")
 
-    window_seconds = hours * 3600
-    window_end = datetime.now(timezone.utc)
-    window_start = window_end - timedelta(hours=hours)
-
     # Per-host window-clamped busy time (each physical machine counted once).
     # This is the source of truth for how loaded each host actually is.
+    # Busy time is the length of the UNION of the host's job intervals, not
+    # their sum: a named runner executes one job at a time, so overlapping
+    # records (e.g. an orphaned job stuck reporting in_progress while the
+    # host serves new jobs) are API artifacts. Summing them pushed saturated
+    # hosts to ~110% utilization; the union caps every host at 100%.
     host_busy_seconds = {}
     for host, jobs in host_jobs.items():
-        busy = 0.0
+        intervals = []
         for j in jobs:
             cs = max(j["start"], window_start)
             ce = min(j["end"], window_end)
             if ce > cs:
-                busy += (ce - cs).total_seconds()
-        host_busy_seconds[host] = busy
+                intervals.append((cs, ce))
+        host_busy_seconds[host] = union_seconds(intervals)
 
     results = []
     for label in sorted(all_labels):
@@ -586,15 +787,16 @@ def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None)
     # Per-job detail (deduped across labels), longest waits first, for the
     # links + status section of the report.
     longest_waits = sorted(all_job_infos, key=lambda j: j["queue_time"], reverse=True)
-    return results, fetch_failure_pct, longest_waits
+    return results, fetch_failure_pct, longest_waits, coverage_hours
 
 
 def format_report(
     results: list[dict],
-    hours: int,
+    hours: float,
     fetch_failure_pct: float = 0.0,
     longest_waits: list = None,
     top_n: int = 20,
+    coverage_hours: float = None,
 ) -> str:
     """One compact summary table — original schema, fixed columns.
 
@@ -606,13 +808,23 @@ def format_report(
     physical hosts, so their utilization now reflects real hardware
     saturation instead of being divided across labels.
     """
+    if coverage_hours is None:
+        coverage_hours = hours
     lines = [
         "# Runner Utilization Report",
         "",
-        f"**Time window:** Last {hours} hours · "
+        f"**Time window:** Last {coverage_hours:.1f} hours · "
         f"**Generated:** {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}",
         "",
     ]
+    if coverage_hours < hours - 0.05:
+        lines.append(
+            f"⚠️ **Coverage warning**: the run listing hit the API safety "
+            f"cap, so only the most recent {coverage_hours:.1f}h of the "
+            f"requested {hours:.0f}h window is covered. All metrics below "
+            f"are computed over the covered {coverage_hours:.1f}h."
+        )
+        lines.append("")
     if fetch_failure_pct > 1.0:
         lines.append(
             f"⚠️ **Data completeness warning**: {fetch_failure_pct:.0f}% of "
@@ -732,17 +944,30 @@ def main():
     parser.add_argument(
         "--hours", type=float, default=24, help="Time window in hours (fractional ok)"
     )
+    parser.add_argument(
+        "--lookback-hours",
+        type=float,
+        default=None,
+        help=(
+            "How far before the window to list runs whose long-queued jobs "
+            "may still execute inside it (default: min(12, hours/2))"
+        ),
+    )
     parser.add_argument(
         "--filter", type=str, help="Filter runner labels (e.g., '5090', 'h200')"
     )
     parser.add_argument("--output", type=str, help="Output file (default: stdout)")
     args = parser.parse_args()
 
-    results, fetch_failure_pct, longest_waits = calculate_utilization(
-        args.repo, args.hours, args.filter
+    results, fetch_failure_pct, longest_waits, coverage_hours = calculate_utilization(
+        args.repo, args.hours, args.filter, lookback_hours=args.lookback_hours
     )
     report = format_report(
-        results, args.hours, fetch_failure_pct, longest_waits=longest_waits
+        results,
+        args.hours,
+        fetch_failure_pct,
+        longest_waits=longest_waits,
+        coverage_hours=coverage_hours,
     )
 
     if args.output:
diff --git a/scripts/ci/utils/test_runner_utilization_report.py b/scripts/ci/utils/test_runner_utilization_report.py
index 8cc692c30..ca4d9b96e 100644
--- a/scripts/ci/utils/test_runner_utilization_report.py
+++ b/scripts/ci/utils/test_runner_utilization_report.py
@@ -191,5 +191,227 @@ class TestStatusAndFormatting(unittest.TestCase):
         self.assertIn("⏳", report)
 
 
+class TestGetWorkflowRuns(unittest.TestCase):
+    """Regression guard for the run-listing truncation bugs.
+
+    Two distinct failure modes are covered:
+    - the original silent 50-page cap (5000 runs) on an UNfiltered listing
+      while ~18k runs fire per busy 24h — fixed by server-side `created`
+      filtering, with honest truncation reporting when a budget is hit;
+    - the GitHub API's 1000-result cap on any `created`-FILTERED listing
+      (page 11 comes back empty despite a larger total_count) — fixed by
+      walking the range's upper bound down past each cap.
+    """
+
+    def _fake_gh_api(self, all_runs):
+        """Emulate the runs-listing API over `all_runs` (newest-first),
+        including the 1000-result-per-filtered-query cap."""
+        calls = []
+
+        def fake(args):
+            path = args[0]
+            calls.append(path)
+            page = int(path.split("&page=")[1].split("&")[0])
+            created = path.split("&created=")[1]
+            if created.startswith(">="):
+                lo, hi = created[2:], None
+            else:
+                lo, hi = created.split("..")
+            matching = [
+                r
+                for r in all_runs
+                if r["created_at"] >= lo and (hi is None or r["created_at"] <= hi)
+            ]
+            offset = (page - 1) * 100
+            # The API serves at most 1000 results per filtered query even
+            # though total_count reports the full match count.
+            page_runs = matching[offset : offset + 100] if offset < 1000 else []
+            return {"total_count": len(matching), "workflow_runs": page_runs}
+
+        return fake, calls
+
+    def _make_runs(self, n):
+        """n runs, newest-first, one per minute."""
+        return [
+            {
+                "id": i,
+                "created_at": _iso(NOW - timedelta(minutes=i + 1)),
+            }
+            for i in range(n)
+        ]
+
+    def _run(self, all_runs, **kw):
+        fake, calls = self._fake_gh_api(all_runs)
+        orig = rur.run_gh_command
+        rur.run_gh_command = fake
+        try:
+            result = rur.get_workflow_runs("o/r", since=NOW - timedelta(hours=24), **kw)
+        finally:
+            rur.run_gh_command = orig
+        return result, calls
+
+    def test_uses_created_filter_and_stops_when_exhausted(self):
+        (runs, truncated), calls = self._run(self._make_runs(130))
+        self.assertEqual(len(runs), 130)
+        self.assertFalse(truncated)
+        self.assertEqual(len(calls), 2)
+        self.assertIn("created=>=", calls[0])
+
+    def test_walks_past_the_1000_result_cap(self):
+        # 1327 runs in-window (the exact live-validation failure): a single
+        # filtered query serves only the newest 1000; the cursor walkdown
+        # must fetch the remaining 327 via a narrowed created range.
+        (runs, truncated), calls = self._run(self._make_runs(1327))
+        self.assertEqual(len(runs), 1327)
+        self.assertEqual(len({r["id"] for r in runs}), 1327)  # deduped
+        self.assertFalse(truncated)
+        # Second sweep queries a bounded range, not the open-ended filter.
+        self.assertTrue(any(".." in c.split("&created=")[1] for c in calls))
+
+    def test_reports_truncation_at_request_budget(self):
+        (runs, truncated), _ = self._run(self._make_runs(1000), max_pages=3)
+        self.assertEqual(len(runs), 300)
+        self.assertTrue(truncated)
+
+
+class TestPreWindowJobFiltering(unittest.TestCase):
+    """Lookback runs bring in jobs that finished before the window started.
+
+    Those jobs must be droppable by comparing their latest activity
+    (busy end or queue end) against window_start — kept only when they
+    touch the window. This mirrors the inline filter in
+    calculate_utilization.
+    """
+
+    def _latest_activity(self, info):
+        return max(t for t in (info["end"], info["queue_end"]) if t is not None)
+
+    def test_finished_pre_window_job_is_droppable(self):
+        window_start = NOW - timedelta(hours=24)
+        info = rur.classify_job(
+            _job(
+                created_at=_iso(NOW - timedelta(hours=30)),
+                started_at=_iso(NOW - timedelta(hours=29)),
+                completed_at=_iso(NOW - timedelta(hours=26)),
+            ),
+            NOW,
+        )
+        self.assertLess(self._latest_activity(info), window_start)
+
+    def test_job_spanning_window_start_is_kept(self):
+        window_start = NOW - timedelta(hours=24)
+        info = rur.classify_job(
+            _job(
+                created_at=_iso(NOW - timedelta(hours=30)),
+                started_at=_iso(NOW - timedelta(hours=26)),
+                completed_at=_iso(NOW - timedelta(hours=20)),
+            ),
+            NOW,
+        )
+        self.assertGreaterEqual(self._latest_activity(info), window_start)
+
+    def test_still_queued_pre_window_job_is_kept(self):
+        # queue_end anchors to `now` for still-queued jobs, so an old run's
+        # still-waiting job always touches the window.
+        window_start = NOW - timedelta(hours=24)
+        info = rur.classify_job(
+            _job(
+                status="queued",
+                runner_name="",
+                created_at=_iso(NOW - timedelta(hours=30)),
+                started_at=_iso(NOW - timedelta(hours=30)),
+                completed_at=None,
+            ),
+            NOW,
+        )
+        self.assertGreaterEqual(self._latest_activity(info), window_start)
+
+
+class TestCarriedOverFingerprint(unittest.TestCase):
+    """Re-run attempts duplicate the completed jobs they did NOT re-run as
+    new records (new id, identical name/runner/timestamps). Verified live:
+    a 2-attempt run listed the same successful job twice under different
+    ids, double-counting its busy time and pass count."""
+
+    def _record(self, **kw):
+        base = _job(
+            started_at=_iso(CREATED + timedelta(minutes=5)),
+            completed_at=_iso(CREATED + timedelta(minutes=25)),
+        )
+        base.update({"run_id": 42, "id": 1})
+        base.update(kw)
+        return base
+
+    def test_carried_over_copy_same_fingerprint(self):
+        a = self._record(id=100, run_attempt=1)
+        b = self._record(id=200, run_attempt=2)  # copy with a new job id
+        self.assertIsNotNone(rur.carried_over_fingerprint(a))
+        self.assertEqual(
+            rur.carried_over_fingerprint(a), rur.carried_over_fingerprint(b)
+        )
+
+    def test_genuine_retry_differs(self):
+        # An actually re-run job has its own execution timestamps.
+        a = self._record(id=100, run_attempt=1)
+        b = self._record(
+            id=200,
+            run_attempt=2,
+            started_at=_iso(CREATED + timedelta(hours=2)),
+            completed_at=_iso(CREATED + timedelta(hours=2, minutes=20)),
+        )
+        self.assertNotEqual(
+            rur.carried_over_fingerprint(a), rur.carried_over_fingerprint(b)
+        )
+
+    def test_non_completed_jobs_not_fingerprinted(self):
+        queued = self._record(status="queued", runner_name="", completed_at=None)
+        self.assertIsNone(rur.carried_over_fingerprint(queued))
+
+
+class TestUnionSeconds(unittest.TestCase):
+    """Per-host busy time must be the union of job intervals, not the sum.
+
+    A named runner executes one job at a time; overlapping records (e.g.
+    an orphaned job stuck in_progress while the host serves new jobs) are
+    API artifacts. Summing them reported saturated hosts at ~110%
+    utilization.
+    """
+
+    def _iv(self, s_min, e_min):
+        return (NOW + timedelta(minutes=s_min), NOW + timedelta(minutes=e_min))
+
+    def test_disjoint_intervals_sum(self):
+        got = rur.union_seconds([self._iv(0, 10), self._iv(20, 30)])
+        self.assertAlmostEqual(got, 20 * 60)
+
+    def test_overlap_counted_once(self):
+        # 0..60 ghost overlapping three real 15-min jobs -> union is 60min,
+        # not 105min.
+        got = rur.union_seconds(
+            [self._iv(0, 60), self._iv(0, 15), self._iv(20, 35), self._iv(40, 55)]
+        )
+        self.assertAlmostEqual(got, 60 * 60)
+
+    def test_unsorted_input_and_touching_edges(self):
+        got = rur.union_seconds([self._iv(10, 20), self._iv(0, 10)])
+        self.assertAlmostEqual(got, 20 * 60)
+
+    def test_empty(self):
+        self.assertEqual(rur.union_seconds([]), 0.0)
+
+
+class TestCoverageWarning(unittest.TestCase):
+    def test_banner_when_coverage_below_requested(self):
+        report = rur.format_report([], 24, 0.0, coverage_hours=9.3)
+        self.assertIn("Coverage warning", report)
+        self.assertIn("9.3h", report)
+
+    def test_no_banner_at_full_coverage(self):
+        report = rur.format_report([], 24, 0.0, coverage_hours=24)
+        self.assertNotIn("Coverage warning", report)
+        report = rur.format_report([], 24, 0.0)  # default: full coverage
+        self.assertNotIn("Coverage warning", report)
+
+
 if __name__ == "__main__":
     unittest.main()