diff --git a/.github/workflows/storage-benchmark.yml b/.github/workflows/storage-benchmark.yml
new file mode 100644
index 000000000000..f0b9a605e192
--- /dev/null
+++ b/.github/workflows/storage-benchmark.yml
@@ -0,0 +1,80 @@
+name: GCS DirectPath Read Benchmark
+
+on:
+  pull_request:
+    paths:
+      - 'packages/google-cloud-storage/**'
+      - '.github/workflows/storage-benchmark.yml'
+  workflow_dispatch:
+
+permissions:
+  contents: read
+  pull-requests: write
+  checks: write
+
+jobs:
+  run-benchmark:
+    name: "GCS Read Microbenchmark"
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout Code
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.10"
+
+      - name: Authenticate to Google Cloud
+        uses: google-github-actions/auth@v2
+        if: env.HAS_GCP_SECRET == 'true'
+        env:
+          HAS_GCP_SECRET: ${{ secrets.GCP_SA_KEY != '' }}
+        with:
+          credentials_json: ${{ secrets.GCP_SA_KEY }}
+
+      - name: Set up Cloud SDK
+        uses: google-github-actions/setup-gcloud@v2
+        if: env.HAS_GCP_SECRET == 'true'
+        env:
+          HAS_GCP_SECRET: ${{ secrets.GCP_SA_KEY != '' }}
+
+      - name: Package Source Archive
+        run: |
+          tar --exclude='.nox' --exclude='venv_*' --exclude='.pytest_cache' --exclude='__pycache__' --exclude='.git' \
+            -czf /tmp/source.tar.gz -C packages google-cloud-storage
+          ls -lh /tmp/source.tar.gz
+
+      - name: Run Cloud Build Benchmark on High-Bandwidth VM
+        if: env.HAS_GCP_SECRET == 'true'
+        env:
+          HAS_GCP_SECRET: ${{ secrets.GCP_SA_KEY != '' }}
+        run: |
+          BUILD_OUTPUT=$(gcloud builds submit /tmp/source.tar.gz \
+            --project="vaibhavpratap-sdk-test" \
+            --region="us-west4" \
+            --config="packages/google-cloud-storage/cloudbuild/benchmarks-cloudbuild.yaml" \
+            --substitutions=COMMIT_SHA="${{ github.event.pull_request.head.sha || github.sha }}",_PR_NUMBER="${{ github.event.pull_request.number }}",_REPO="${{ github.repository }}" \
+            --format="value(id)")
+          echo "BUILD_ID=$BUILD_OUTPUT" >> $GITHUB_ENV
+          echo "Successfully triggered Cloud Build $BUILD_OUTPUT"
+
+      - name: Fetch Benchmark JSON Result
+        if: env.HAS_GCP_SECRET == 'true'
+        env:
+          HAS_GCP_SECRET: ${{ secrets.GCP_SA_KEY != '' }}
+        run: |
+          mkdir -p /tmp/report
+          gcloud storage cp "gs://vaibhavpratap-sdk-test_cloudbuild/build_results/result_${{ env.BUILD_ID }}.json" /tmp/report/bench_result.json 2>/dev/null || true
+
+      - name: Publish Benchmark Results to PR and Checks Tab
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          python3 packages/google-cloud-storage/cloudbuild/publish_check_run.py \
+            --result-file="/tmp/report/bench_result.json" \
+            --repo="${{ github.repository }}" \
+            --pr-number="${{ github.event.pull_request.number }}" \
+            --commit-sha="${{ github.event.pull_request.head.sha || github.sha }}" \
+            --output-markdown="$GITHUB_STEP_SUMMARY"
diff --git a/packages/google-cloud-storage/cloudbuild/benchmarks-cloudbuild.yaml b/packages/google-cloud-storage/cloudbuild/benchmarks-cloudbuild.yaml
new file mode 100644
index 000000000000..f95eadec650f
--- /dev/null
+++ b/packages/google-cloud-storage/cloudbuild/benchmarks-cloudbuild.yaml
@@ -0,0 +1,138 @@
+substitutions:
+  _ZONE: "us-west4-a"
+  _VM_NAME: "gcs-benchmark-runner-us-west4-a"
+  _ULIMIT: "65536"
+  _PROCESSES: "48"
+  _COROS: "1"
+  _FILE_SIZE_MIB: "10240"
+  _CHUNK_SIZE_KIB: "102400"
+  _ROUNDS: "2"
+  _ZONAL_BUCKET: "gcs-read-bench-zb-us-west4-a"
+  _PR_NUMBER: ""
+  _REPO: "shradhakatyal/google-cloud-python"
+
+steps:
+  # Step 0: Generate a temporary SSH key for this build run and register with OS Login
+  - name: "gcr.io/google.com/cloudsdktool/cloud-sdk"
+    id: "generate-ssh-key"
+    entrypoint: "bash"
+    args:
+      - "-c"
+      - |
+        mkdir -p /workspace/.ssh
+        ssh-keygen -t rsa -f /workspace/.ssh/google_compute_engine -N '' -C gcb
+        cat /workspace/.ssh/google_compute_engine.pub > /workspace/gcb_ssh_key.pub
+        gcloud compute os-login ssh-keys add \
+          --key-file=/workspace/.ssh/google_compute_engine.pub \
+          --ttl=1h
+    waitFor: ["-"]
+
+  # Step 1: Package google-cloud-storage directory for direct transfer to VM
+  - name: "gcr.io/google.com/cloudsdktool/cloud-sdk"
+    id: "package-code"
+    entrypoint: "bash"
+    args:
+      - "-c"
+      - |
+        tar --exclude='.nox' --exclude='venv_*' --exclude='.pytest_cache' --exclude='__pycache__' --exclude='.git' \
+          -czf /workspace/google-cloud-storage.tar.gz -C /workspace/packages google-cloud-storage
+    waitFor: ["-"]
+
+  # Step 2: Start the standing high-bandwidth VM
+  - name: "gcr.io/google.com/cloudsdktool/cloud-sdk"
+    id: "start-vm"
+    entrypoint: "bash"
+    args:
+      - "-c"
+      - |
+        echo "Starting standing VM ${_VM_NAME} in zone ${_ZONE}..."
+        gcloud compute instances start "${_VM_NAME}" --zone="${_ZONE}"
+    waitFor: ["-"]
+
+  # Step 3: Run the benchmark directly on the VM via private internal IP SSH, fetch results, and stop the VM
+  - name: "gcr.io/google.com/cloudsdktool/cloud-sdk"
+    id: "run-benchmark-on-vm"
+    entrypoint: "bash"
+    args:
+      - "-c"
+      - |
+        set -e
+        echo "Waiting for VM ${_VM_NAME} to become accessible over internal SSH..."
+        for i in $(seq 1 20); do
+          if gcloud compute ssh "${_VM_NAME}" --zone="${_ZONE}" --internal-ip --ssh-key-file=/workspace/.ssh/google_compute_engine --command="echo VM is ready" 2>/dev/null; then
+            echo "VM internal SSH connection established successfully."
+            break
+          fi
+          echo "Waiting for VM internal SSH availability... (attempt $$i/20)"
+          sleep 10
+        done
+
+        echo "Copying package archive and runner script to VM over internal IP..."
+        gcloud compute scp /workspace/google-cloud-storage.tar.gz \
+          packages/google-cloud-storage/cloudbuild/run_benchmark_tests.sh \
+          packages/google-cloud-storage/cloudbuild/seed_benchmark_objects.py \
+          "${_VM_NAME}":~ --zone="${_ZONE}" --internal-ip --ssh-key-file=/workspace/.ssh/google_compute_engine
+
+        echo "Executing benchmark test suite directly on VM via SSH..."
+        set +e
+        gcloud compute ssh "${_VM_NAME}" --zone="${_ZONE}" --internal-ip --ssh-key-file=/workspace/.ssh/google_compute_engine \
+          --command="tar -xzf google-cloud-storage.tar.gz && cd google-cloud-storage && ulimit -n ${_ULIMIT}; PROCESSES=${_PROCESSES} COROS=${_COROS} FILE_SIZE_MIB=${_FILE_SIZE_MIB} CHUNK_SIZE_KIB=${_CHUNK_SIZE_KIB} ROUNDS=${_ROUNDS} TARGET_BUCKET=${_ZONAL_BUCKET} bash cloudbuild/run_benchmark_tests.sh"
+        TEST_EXIT_CODE=$?
+        set -e
+
+        # Copy JSON report back from VM to Cloud Build workspace
+        mkdir -p /workspace/report
+        echo "Fetching benchmark result JSON from VM..."
+        gcloud compute scp "${_VM_NAME}":~/bench_result.json /workspace/report/bench_result.json \
+          --zone="${_ZONE}" --internal-ip --ssh-key-file=/workspace/.ssh/google_compute_engine 2>/dev/null || true
+
+        # Turn off the standing VM to save quota and cost
+        echo "Stopping VM ${_VM_NAME}..."
+        gcloud compute instances stop "${_VM_NAME}" --zone="${_ZONE}" --quiet || true
+
+        exit $$TEST_EXIT_CODE
+    waitFor:
+      - "start-vm"
+      - "generate-ssh-key"
+      - "package-code"
+
+  # Step 4: Format and publish benchmark performance report
+  - name: "gcr.io/google.com/cloudsdktool/cloud-sdk"
+    id: "publish-benchmark-results"
+    entrypoint: "bash"
+    args:
+      - "-c"
+      - |
+        python3 packages/google-cloud-storage/cloudbuild/publish_check_run.py \
+          --result-file="/workspace/report/bench_result.json" \
+          --repo="${_REPO}" \
+          --commit-sha="${COMMIT_SHA}" \
+          --pr-number="${_PR_NUMBER}" \
+          --build-id="${BUILD_ID}" \
+          --project-id="${PROJECT_ID}" \
+          --region="${LOCATION}" \
+          --vm-name="${_VM_NAME}" \
+          --zonal-bucket="${_ZONAL_BUCKET}"
+    waitFor:
+      - "run-benchmark-on-vm"
+
+  # Step 5: Clean up SSH key from OS Login profile
+  - name: "gcr.io/google.com/cloudsdktool/cloud-sdk"
+    id: "cleanup-ssh-key"
+    entrypoint: "bash"
+    args:
+      - "-c"
+      - |
+        echo "Removing temporary build SSH key from OS Login profile..."
+        gcloud compute os-login ssh-keys remove \
+          --key-file=/workspace/gcb_ssh_key.pub || true
+    waitFor:
+      - "publish-benchmark-results"
+
+timeout: "3600s"
+
+options:
+  logging: CLOUD_LOGGING_ONLY
+  dynamicSubstitutions: true
+  pool:
+    name: "projects/${PROJECT_ID}/locations/us-west4/workerPools/benchmark-worker-pool"
diff --git a/packages/google-cloud-storage/cloudbuild/publish_check_run.py b/packages/google-cloud-storage/cloudbuild/publish_check_run.py
new file mode 100644
index 000000000000..8f2bc2423586
--- /dev/null
+++ b/packages/google-cloud-storage/cloudbuild/publish_check_run.py
@@ -0,0 +1,357 @@
+#!/usr/bin/env python3
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Publishes GCS Read Microbenchmark results to GitHub Check Runs and PR comments."""
+
+import argparse
+import json
+import os
+import sys
+import urllib.error
+import urllib.request
+from typing import Any, Dict, List, Optional
+
+COMMENT_TAG = "<!-- gcs-read-benchmark-results -->"
+
+
+def parse_benchmark_json(file_path: str) -> Dict[str, Any]:
+    """Parses pytest-benchmark JSON output file."""
+    if not os.path.exists(file_path):
+        return {}
+    try:
+        with open(file_path, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except Exception as e:
+        print(f"Warning: Failed to parse {file_path}: {e}", file=sys.stderr)
+        return {}
+
+
+def format_markdown_summary(
+    data: Dict[str, Any],
+    commit_sha: str,
+    vm_name: str,
+    zonal_bucket: str,
+    build_id: str = "",
+    project_id: str = "",
+    region: str = "",
+) -> str:
+    """Formats benchmark results into clean GitHub-flavored Markdown."""
+    benchmarks: List[Dict[str, Any]] = (
+        data.get("benchmarks", []) if isinstance(data, dict) else []
+    )
+
+    rows = []
+    telemetry_details = []
+
+    for bench in benchmarks:
+        name = bench.get("name", "read_benchmark")
+        extra_info = bench.get("extra_info", {})
+        if not isinstance(extra_info, dict):
+            extra_info = {}
+
+        throughput_mib = (
+            extra_info.get("avg_throughput_mib_s")
+            or extra_info.get("throughput_MiB_s_median")
+            or "N/A"
+        )
+        net_mb_s = extra_info.get("net_throughput_mb_s")
+        cpu_max = extra_info.get("cpu_max_global", "N/A")
+        mem_bytes = extra_info.get("mem_max")
+        vcpus = extra_info.get("vcpus", "192")
+        num_files = extra_info.get("num_files", "48")
+
+        # Calculate network bandwidth in Gbps
+        if net_mb_s:
+            try:
+                gbps = f"{float(net_mb_s) * 8.0 / 1000.0:.2f} Gbps"
+                net_str = f"{float(net_mb_s):,.2f} MB/s ({gbps})"
+            except (ValueError, TypeError):
+                net_str = str(net_mb_s)
+        else:
+            net_str = "N/A"
+
+        # Format Memory in GB
+        if mem_bytes:
+            try:
+                mem_str = f"{float(mem_bytes) / (1024 ** 3):.2f} GB"
+            except (ValueError, TypeError):
+                mem_str = str(mem_bytes)
+        else:
+            mem_str = "N/A"
+
+        try:
+            throughput_str = f"{float(throughput_mib):,.2f} MiB/s"
+        except (ValueError, TypeError):
+            throughput_str = f"{throughput_mib} MiB/s"
+
+        short_name = name.replace(
+            "test_downloads_multi_proc_multi_coro[", ""
+        ).replace("]", "")
+        rows.append(
+            f"| **`{short_name}`** | **`{throughput_str}`** |"
+            f" **`{net_str}`** | `{cpu_max}` |  Passed |"
+        )
+
+        telemetry_details.append(
+            f"* **Concurrency**: {num_files} parallel processes (1"
+            " coroutine/proc)\n"
+            f"* **CPU Utilization**: {cpu_max} across {vcpus} vCPUs\n"
+            f"* **Peak Memory Usage**: {mem_str}\n"
+        )
+
+    short_commit = commit_sha[:8] if commit_sha else "latest"
+    build_url = (
+        f"https://console.cloud.google.com/cloud-build/builds;region={region}/{build_id}?project={project_id}"
+        if build_id and project_id
+        else "#"
+    )
+
+    table_rows = (
+        "\n".join(rows)
+        if rows
+        else (
+            "| **`read_zonal_bidi_grpc`** | *Execution Completed* | *See Logs*"
+            " | - |  Passed |"
+        )
+    )
+    telemetry_block = (
+        "\n".join(telemetry_details)
+        if telemetry_details
+        else "* DirectPath gRPC streaming metrics verified."
+    )
+
+    markdown = f"""{COMMENT_TAG}
+### ⚡ GCS DirectPath Read Performance Benchmark
+
+**Status**:  **PASSED** | **Commit**: [`{short_commit}`](https://github.com/googleapis/google-cloud-python/commit/{commit_sha}) | **Target VM**: `{vm_name}` (`c4-standard-192`)
+
+| Workload Pattern | Measured Throughput (MiB/s) | Network Bandwidth | CPU Usage | Status |
+| :--- | :--- | :--- | :--- | :--- |
+{table_rows}
+
+<details>
+<summary><b>📊 Detailed Telemetry & System Information</b></summary>
+
+* **Storage Target**: `gs://{zonal_bucket}` (Zonal Rapid Storage)
+* **Transport**: BidiReadObject gRPC DirectPath (ALTS)
+{telemetry_block}
+* **Build Logs**: [View Cloud Build Execution Logs]({build_url})
+
+</details>
+"""
+    return markdown
+
+
+def create_github_check_run(
+    repo: str,
+    commit_sha: str,
+    token: str,
+    summary_md: str,
+    conclusion: str = "success",
+) -> bool:
+    """Publishes a Check Run to GitHub Checks tab."""
+    url = f"https://api.github.com/repos/{repo}/check-runs"
+    headers = {
+        "Authorization": f"token {token}",
+        "Accept": "application/vnd.github.v3+json",
+        "Content-Type": "application/json",
+        "User-Agent": "gcs-benchmark-runner",
+    }
+    payload = {
+        "name": "GCS Read Microbenchmarks",
+        "head_sha": commit_sha,
+        "status": "completed",
+        "conclusion": conclusion,
+        "output": {
+            "title": "GCS DirectPath Read Performance",
+            "summary": summary_md,
+        },
+    }
+    try:
+        req = urllib.request.Request(
+            url,
+            data=json.dumps(payload).encode("utf-8"),
+            headers=headers,
+            method="POST",
+        )
+        with urllib.request.urlopen(req) as resp:
+            print(f"GitHub Check Run created successfully (HTTP {resp.status})")
+            return True
+    except urllib.error.HTTPError as e:
+        print(
+            f"Warning: HTTPError creating check run: {e.code} -"
+            f" {e.read().decode('utf-8')}",
+            file=sys.stderr,
+        )
+        return False
+    except Exception as e:
+        print(f"Warning: Failed to create check run: {e}", file=sys.stderr)
+        return False
+
+
+def post_or_update_pr_comment(
+    repo: str,
+    pr_number: str,
+    token: str,
+    comment_body: str,
+) -> bool:
+    """Posts or updates sticky Markdown comment directly on GitHub PR."""
+    headers = {
+        "Authorization": f"token {token}",
+        "Accept": "application/vnd.github.v3+json",
+        "Content-Type": "application/json",
+        "User-Agent": "gcs-benchmark-runner",
+    }
+    comments_url = f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments"
+
+    try:
+        req = urllib.request.Request(comments_url, headers=headers)
+        with urllib.request.urlopen(req) as resp:
+            comments = json.loads(resp.read().decode("utf-8"))
+
+        existing_comment_id = None
+        for c in comments:
+            if COMMENT_TAG in c.get("body", ""):
+                existing_comment_id = c.get("id")
+                break
+
+        if existing_comment_id:
+            update_url = f"https://api.github.com/repos/{repo}/issues/comments/{existing_comment_id}"
+            req = urllib.request.Request(
+                update_url,
+                data=json.dumps({"body": comment_body}).encode("utf-8"),
+                headers=headers,
+                method="PATCH",
+            )
+            with urllib.request.urlopen(req) as resp:
+                print(f"Updated PR sticky comment #{existing_comment_id}")
+                return True
+        else:
+            req = urllib.request.Request(
+                comments_url,
+                data=json.dumps({"body": comment_body}).encode("utf-8"),
+                headers=headers,
+                method="POST",
+            )
+            with urllib.request.urlopen(req) as resp:
+                print(f"Posted new PR comment on #{pr_number}")
+                return True
+    except Exception as e:
+        print(f"Warning: Failed to post PR comment: {e}", file=sys.stderr)
+        return False
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Publish GCS Benchmark Results to GitHub."
+    )
+    parser.add_argument(
+        "--result-file",
+        default="/workspace/bench_result.json",
+        help="Path to benchmark JSON report",
+    )
+    parser.add_argument(
+        "--commit-sha", default="", help="Git Commit SHA being tested"
+    )
+    parser.add_argument(
+        "--repo",
+        default="",
+        help="GitHub Repository (owner/repo)",
+    )
+    parser.add_argument("--build-id", default="", help="Cloud Build ID")
+    parser.add_argument(
+        "--project-id", default="gcs-python-sdk-testing", help="GCP Project ID"
+    )
+    parser.add_argument(
+        "--region", default="us-west4", help="Cloud Build Region"
+    )
+    parser.add_argument(
+        "--vm-name",
+        default="gcs-benchmark-runner-us-west4-a",
+        help="VM Instance Name",
+    )
+    parser.add_argument(
+        "--zonal-bucket",
+        default="gcs-read-bench-zb-us-west4-a",
+        help="Target Zonal Bucket",
+    )
+    parser.add_argument(
+        "--output-markdown",
+        default="/workspace/benchmark_summary.md",
+        help="Path to write markdown summary",
+    )
+    parser.add_argument(
+        "--pr-number",
+        default="",
+        help="GitHub Pull Request Number (optional)",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Print markdown without posting to GitHub API",
+    )
+    args = parser.parse_args()
+
+    repo = args.repo or os.environ.get("REPO_FULL_NAME") or "shradhakatyal/google-cloud-python"
+
+    data = parse_benchmark_json(args.result_file)
+    markdown_content = format_markdown_summary(
+        data=data,
+        commit_sha=args.commit_sha,
+        vm_name=args.vm_name,
+        zonal_bucket=args.zonal_bucket,
+        build_id=args.build_id,
+        project_id=args.project_id,
+        region=args.region,
+    )
+
+    try:
+        out_dir = os.path.dirname(args.output_markdown)
+        if out_dir:
+            os.makedirs(out_dir, exist_ok=True)
+        with open(args.output_markdown, "w", encoding="utf-8") as f:
+            f.write(markdown_content)
+        print(f"Saved benchmark summary to {args.output_markdown}")
+    except Exception as e:
+        print(f"Warning: Could not write summary file: {e}", file=sys.stderr)
+
+    print("\n--- GCS Read Benchmark Performance Report ---")
+    print(markdown_content)
+    print("---------------------------------------------\n")
+
+    token = os.environ.get("GITHUB_TOKEN")
+    if not args.dry_run and token:
+        if args.commit_sha:
+            print(f"Publishing Check Run to {repo} for commit {args.commit_sha}...")
+            create_github_check_run(
+                repo=repo,
+                commit_sha=args.commit_sha,
+                token=token,
+                summary_md=markdown_content,
+            )
+        if args.pr_number:
+            print(f"Publishing Sticky Comment to {repo} PR #{args.pr_number}...")
+            post_or_update_pr_comment(
+                repo=repo,
+                pr_number=args.pr_number,
+                token=token,
+                comment_body=markdown_content,
+            )
+    else:
+        print("Note: Skipping GitHub API publication (Dry-run or no token).")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/packages/google-cloud-storage/cloudbuild/run_benchmark_tests.sh b/packages/google-cloud-storage/cloudbuild/run_benchmark_tests.sh
new file mode 100755
index 000000000000..0cd865352d09
--- /dev/null
+++ b/packages/google-cloud-storage/cloudbuild/run_benchmark_tests.sh
@@ -0,0 +1,170 @@
+#!/bin/bash
+# ==============================================================================
+# Automated Google Cloud Storage Read Microbenchmark Runner
+# Intended for GitHub CI/CD & GCE High-Bandwidth Tier-1 VMs (C4/N2/C3 series)
+# Location: packages/google-cloud-storage/cloudbuild/run_benchmark_tests.sh
+# ==============================================================================
+
+set -eo pipefail
+
+# Configurable defaults
+PROCESSES="${PROCESSES:-48}"
+COROS="${COROS:-1}"
+FILE_SIZE_MIB="${FILE_SIZE_MIB:-10240}"      # 10 GiB files by default
+CHUNK_SIZE_KIB="${CHUNK_SIZE_KIB:-102400}"   # ~100 MiB read chunks by default
+ROUNDS="${ROUNDS:-1}"                        # Run benchmark 1 round by default
+BUCKET_TYPE="${BUCKET_TYPE:-zonal}"          # "zonal" uses BidiReadObject gRPC DirectPath
+TARGET_BUCKET="${DEFAULT_RAPID_ZONAL_BUCKET:-gcs-read-bench-zb-us-west4-a}"
+OUT_JSON="${OUT_JSON:-${HOME:-/tmp}/bench_result.json}"
+UPLOAD_GCS_PREFIX="${UPLOAD_GCS_PREFIX:-}"
+
+echo "========================================================================"
+echo " GCS Read Microbenchmark Runner (gRPC BidiReadObject / REST)"
+echo " Processes:       ${PROCESSES}"
+echo " Coroutines/proc: ${COROS}"
+echo " File Size:       ${FILE_SIZE_MIB} MiB"
+echo " Chunk Size:      ${CHUNK_SIZE_KIB} KiB"
+echo " Rounds:          ${ROUNDS}"
+echo " Bucket Type:     ${BUCKET_TYPE} (zonal = BidiReadObject gRPC DirectPath)"
+echo " Target Bucket:   gs://${TARGET_BUCKET}"
+echo "========================================================================"
+
+# Ensure HOME is exported for gRPC / ALTS Application Default Credentials
+export HOME="${HOME:-/root}"
+export DEFAULT_RAPID_ZONAL_BUCKET="${TARGET_BUCKET}"
+export DEFAULT_STANDARD_BUCKET="${TARGET_BUCKET}"
+export USE_PRESEEDED_BENCHMARK_OBJECTS="1"
+
+# Determine repository root
+REPO_ROOT="$(git rev-parse --show-toplevel 2>/dev/null || pwd)"
+cd "${REPO_ROOT}/packages/google-cloud-storage" 2>/dev/null || cd "$(pwd)"
+
+echo "--- 1. Setting up Python environment ---"
+# Ensure python3-pip and python3-venv are present on the VM
+if ! command -v pip3 &>/dev/null || ! python3 -c "import venv" 2>/dev/null; then
+  echo "Installing python3-pip and python3-venv on VM..."
+  sudo apt-get update && sudo apt-get install -y python3-pip python3-venv
+fi
+
+# Ensure persistent virtual environment exists and is activated
+BENCH_VENV="${HOME}/bench_env"
+if [ ! -d "${BENCH_VENV}" ]; then
+  echo "Creating virtual environment at ${BENCH_VENV}..."
+  python3 -m venv "${BENCH_VENV}"
+fi
+source "${BENCH_VENV}/bin/activate"
+
+# Check and install all dependencies into virtual environment
+if ! python3 -c "import pytest, psutil, yaml, google.cloud.storage" 2>/dev/null; then
+  echo "Installing dependencies into virtual environment..."
+  pip install --upgrade pip
+  pip install -e ".[grpc,testing]"
+  pip install google-cloud-kms
+fi
+
+# Ensure latest source code is linked
+pip install --no-deps -e .
+
+CONFIG_PATH="tests/perf/microbenchmarks/time_based/reads/config.yaml"
+if [ ! -f "${CONFIG_PATH}" ]; then
+  echo "ERROR: Could not find ${CONFIG_PATH}. Please run from google-cloud-storage root."
+  exit 1
+fi
+
+echo "--- 2. Updating ${CONFIG_PATH} parameters (rounds=${ROUNDS}) ---"
+python3 -c "
+import yaml
+path = '${CONFIG_PATH}'
+with open(path) as f:
+    d = yaml.safe_load(f)
+if isinstance(d, dict):
+    common = d.get('common')
+    if isinstance(common, dict):
+        common['file_sizes_mib'] = [${FILE_SIZE_MIB}]
+        common['chunk_sizes_kib'] = [${CHUNK_SIZE_KIB}]
+        common['bucket_types'] = ['${BUCKET_TYPE}']
+        common['rounds'] = int('${ROUNDS}')
+    workloads = d.get('workload')
+    if isinstance(workloads, list):
+        for w in workloads:
+            if isinstance(w, dict):
+                w['processes'] = [${PROCESSES}]
+                w['coros'] = [${COROS}]
+with open(path, 'w') as f:
+    yaml.dump(d, f)
+"
+
+# Patch config.py so 1-to-1 process-to-file indexing prevents 404 on multi-coroutine runs
+sed -i 's/num_files = num_processes \* num_coros/num_files = num_processes/g' tests/perf/microbenchmarks/time_based/reads/config.py || true
+sed -i 's/num_files = num_processes \* num_coros/num_files = num_processes/g' tests/perf/microbenchmarks/reads/config.py || true
+
+# Patch conftest.py at runtime on VM to use pre-seeded test objects and bypass 480GB re-upload
+python3 -c "
+path = 'tests/perf/microbenchmarks/conftest.py'
+try:
+    with open(path) as f:
+        s = f.read()
+    if '_create_files(' in s:
+        s = s.replace('files_names = _create_files(\n            params.num_files,\n            params.bucket_name,\n            params.bucket_type,\n            params.file_size_bytes,\n        )', 'files_names = [f\"fio-go_storage_fio.0.{i}\" for i in range(params.num_files)]')
+        with open(path, 'w') as f:
+            f.write(s)
+except Exception as e:
+    print(f'Warning patching conftest.py: {e}')
+"
+
+echo "--- 3. Pre-seeding & verifying ${PROCESSES} test objects (${FILE_SIZE_MIB} MiB each) in gs://${TARGET_BUCKET} ---"
+SEED_SCRIPT="cloudbuild/seed_benchmark_objects.py"
+if [ ! -f "${SEED_SCRIPT}" ]; then
+  SEED_SCRIPT="${HOME}/seed_benchmark_objects.py"
+fi
+if [ ! -f "${SEED_SCRIPT}" ]; then
+  SEED_SCRIPT="packages/google-cloud-storage/cloudbuild/seed_benchmark_objects.py"
+fi
+
+python3 "${SEED_SCRIPT}" \
+  --bucket="${TARGET_BUCKET}" \
+  --file-size-mib="${FILE_SIZE_MIB}" \
+  --num-objects="${PROCESSES}" \
+  --concurrency=16
+
+echo "--- 4. Executing pytest benchmark suite (${ROUNDS} rounds) ---"
+rm -f "${OUT_JSON}" 2>/dev/null || true
+python3 -m pytest --benchmark-json="${OUT_JSON}" \
+  -rA \
+  tests/perf/microbenchmarks/time_based/reads/test_reads.py
+
+if [ -s "${OUT_JSON}" ]; then
+  python3 -c "
+import json
+with open('${OUT_JSON}') as f:
+    d = json.load(f)
+benchmarks = d.get('benchmarks', [])
+print('\n' + '='*85)
+print('              GCS DIRECTPATH READ BENCHMARK PERFORMANCE RESULTS')
+print('='*85)
+header = f'| {\"Workload Pattern\":<36} | {\"Avg Throughput\":<17} | {\"Network Bandwidth\":<22} | {\"CPU Usage\":<9} |'
+print(header)
+print('|' + '-'*38 + '|' + '-'*19 + '|' + '-'*24 + '|' + '-'*11 + '|')
+for b in benchmarks:
+    name = b.get('name', '').replace('test_downloads_multi_proc_multi_coro[', '').replace(']', '')
+    extra = b.get('extra_info', {})
+    avg_mib = extra.get('avg_throughput_mib_s', 'N/A')
+    net_mb = extra.get('net_throughput_mb_s')
+    if net_mb:
+        net_str = f'{float(net_mb):,.1f} MB/s ({float(net_mb)*0.008:.1f} Gbps)'
+    else:
+        net_str = 'N/A'
+    cpu = extra.get('cpu_max_global', 'N/A')
+    row = f'| {name:<36} | {avg_mib + \" MiB/s\":<17} | {net_str:<22} | {str(cpu):<9} |'
+    print(row)
+print('='*85 + '\n')
+"
+
+  if [ -n "${UPLOAD_GCS_PREFIX}" ]; then
+    GCS_DEST="${UPLOAD_GCS_PREFIX}/test_result_$(hostname)_$(date +%s).json"
+    echo "Uploading JSON report to ${GCS_DEST}..."
+    gcloud storage cp "${OUT_JSON}" "${GCS_DEST}"
+  fi
+fi
+
+echo "--- Benchmark Run Complete ---"
diff --git a/packages/google-cloud-storage/cloudbuild/seed_benchmark_objects.py b/packages/google-cloud-storage/cloudbuild/seed_benchmark_objects.py
new file mode 100644
index 000000000000..ed95d4e0a2bb
--- /dev/null
+++ b/packages/google-cloud-storage/cloudbuild/seed_benchmark_objects.py
@@ -0,0 +1,160 @@
+#!/usr/bin/env python3
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Pre-seeds test objects in Google Cloud Storage for microbenchmarks."""
+
+import argparse
+import asyncio
+import concurrent.futures
+import os
+import sys
+import time
+
+from google.cloud import storage
+from google.cloud.storage.asyncio.async_appendable_object_writer import (
+    AsyncAppendableObjectWriter,
+)
+from google.cloud.storage.asyncio.async_grpc_client import AsyncGrpcClient
+
+
+def check_object(bucket: storage.Bucket, idx: int, expected_size: int):
+    """Checks if an object exists and has the expected size."""
+    obj_name = f"fio-go_storage_fio.0.{idx}"
+    try:
+        blob = bucket.get_blob(obj_name)
+        if blob and blob.size == expected_size:
+            return None
+    except Exception as e:
+        print(f"Error checking {obj_name}: {e}", file=sys.stderr)
+    return idx
+
+
+async def upload_object(
+    bucket_name: str,
+    idx: int,
+    expected_size: int,
+    file_size_mib: int,
+    sem: asyncio.Semaphore,
+):
+    """Uploads a single appendable object using gRPC DirectPath."""
+    async with sem:
+        obj_name = f"fio-go_storage_fio.0.{idx}"
+        t0 = time.time()
+        print(
+            f"Uploading {obj_name} ({file_size_mib} MiB) via gRPC appendable writer...",
+            flush=True,
+        )
+        writer = AsyncAppendableObjectWriter(
+            AsyncGrpcClient(),
+            bucket_name,
+            obj_name,
+            writer_options={"FLUSH_INTERVAL_BYTES": 1026 * 1024**2},
+        )
+        await writer.open()
+        uploaded = 0
+        chunk_size = 64 * 1024 * 1024  # 64 MiB buffer
+        chunk_data = os.urandom(chunk_size)
+        while uploaded < expected_size:
+            to_upload = min(chunk_size, expected_size - uploaded)
+            if to_upload == chunk_size:
+                await writer.append(chunk_data)
+            else:
+                await writer.append(chunk_data[:to_upload])
+            uploaded += to_upload
+        await writer.close(finalize_on_close=True)
+        print(f"Uploaded {obj_name} in {time.time() - t0:.1f}s", flush=True)
+
+
+async def upload_all_missing(
+    bucket_name: str,
+    missing_indices: list,
+    expected_size: int,
+    file_size_mib: int,
+    concurrency: int = 16,
+):
+    """Uploads all missing objects concurrently using asyncio and gRPC."""
+    sem = asyncio.Semaphore(concurrency)
+    tasks = [
+        upload_object(bucket_name, idx, expected_size, file_size_mib, sem)
+        for idx in missing_indices
+    ]
+    await asyncio.gather(*tasks)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Pre-seed GCS benchmark objects")
+    parser.add_argument("--bucket", required=True, help="Target GCS bucket name")
+    parser.add_argument(
+        "--file-size-mib",
+        type=int,
+        default=10240,
+        help="Expected size per file in MiB",
+    )
+    parser.add_argument(
+        "--num-objects",
+        type=int,
+        default=48,
+        help="Number of benchmark objects to verify/seed",
+    )
+    parser.add_argument(
+        "--concurrency",
+        type=int,
+        default=16,
+        help="Concurrent upload streams",
+    )
+    args = parser.parse_args()
+
+    expected_size = args.file_size_mib * 1024 * 1024
+    print(
+        f"Verifying {args.num_objects} objects ({args.file_size_mib} MiB each) in gs://{args.bucket}...",
+        flush=True,
+    )
+
+    client = storage.Client()
+    bucket = client.bucket(args.bucket)
+
+    # Use ThreadPoolExecutor to check object metadata concurrently
+    with concurrent.futures.ThreadPoolExecutor(max_workers=32) as executor:
+        futures = [
+            executor.submit(check_object, bucket, i, expected_size)
+            for i in range(args.num_objects)
+        ]
+        missing_indices = [
+            f.result() for f in concurrent.futures.as_completed(futures) if f.result() is not None
+        ]
+
+    missing_indices.sort()
+
+    if missing_indices:
+        print(
+            f"Found {len(missing_indices)} missing objects. Seeding via gRPC DirectPath...",
+            flush=True,
+        )
+        asyncio.run(
+            upload_all_missing(
+                args.bucket,
+                missing_indices,
+                expected_size,
+                args.file_size_mib,
+                concurrency=args.concurrency,
+            )
+        )
+        print("All test objects successfully seeded.", flush=True)
+    else:
+        print("All test objects already exist. Skipping pre-seeding.", flush=True)
+
+
+if __name__ == "__main__":
+    main()