diff --git a/.github/workflows/auto_draft_release.yaml b/.github/workflows/auto_draft_release.yaml
index 2882efb6..f7c93c11 100644
--- a/.github/workflows/auto_draft_release.yaml
+++ b/.github/workflows/auto_draft_release.yaml
@@ -6,13 +6,14 @@ on:
       - main
 
 jobs:
-  draft-release:
+  check-bump:
     runs-on: ubuntu-latest
-    permissions:
-      contents: write
+    outputs:
+      is_bump: ${{ steps.check-bump.outputs.is_bump }}
+      version: ${{ steps.check-bump.outputs.version }}
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
 
       - name: Check if version bump commit
         id: check-bump
@@ -29,14 +30,65 @@ jobs:
             echo "Not a version bump commit."
           fi
 
-      - name: Create Draft Release
-        if: steps.check-bump.outputs.is_bump == 'true'
-        uses: softprops/action-gh-release@v2
+  gcp-integration-test:
+    needs: check-bump
+    if: needs.check-bump.outputs.is_bump == 'true'
+    runs-on: ubuntu-latest
+    permissions:
+      contents: 'read'
+      id-token: 'write'
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v5
+
+      - name: Authenticate to GCP
+        uses: google-github-actions/auth@v2
         with:
-          tag_name: v${{ steps.check-bump.outputs.version }}
-          name: v${{ steps.check-bump.outputs.version }}
-          draft: true
-          prerelease: ${{ contains(steps.check-bump.outputs.version, 'rc') }}
-          generate_release_notes: true
+          workload_identity_provider: ${{ secrets.GCP_WIF_PROVIDER }}
+          service_account: ${{ secrets.GCP_WIF_SA_EMAIL }}
+
+      - name: Set up GCP Cloud SDK
+        uses: google-github-actions/setup-gcloud@v2
+        with:
+          project_id: datcom-ci
+
+      - name: Install uv and Set up Python
+        uses: astral-sh/setup-uv@v6
+        with:
+          python-version: "3.12"
+
+      - name: Install Dependencies
+        run: uv sync --locked
+
+      - name: Run GCP Integration Test
         env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          DC_API_KEY: ${{ secrets.DC_API_KEY }}
+        run: |
+          TAG="${{ needs.check-bump.outputs.version }}"
+          
+          uv run python3 tests/datacommons-integration-tests/gcp_test_runner.py \
+            --project-id datcom-ci \
+            --dcp-version "$TAG" \
+            --dc-api-key "$DC_API_KEY"
+
+  hermetic-integration-test:
+    needs: check-bump
+    if: needs.check-bump.outputs.is_bump == 'true'
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v5
+
+      - name: Install uv and Set up Python
+        uses: astral-sh/setup-uv@v6
+        with:
+          python-version: "3.12"
+
+      - name: Install Dependencies
+        run: uv sync --locked
+
+      - name: Run hermetic integration test
+        run: uv run pytest -s tests/datacommons-integration-tests/run_local_integration_test.py
+
+
+
diff --git a/.github/workflows/gcp_integration_test.yaml b/.github/workflows/gcp_integration_test.yaml
new file mode 100644
index 00000000..fd0b6dd9
--- /dev/null
+++ b/.github/workflows/gcp_integration_test.yaml
@@ -0,0 +1,92 @@
+name: GCP Sandbox Integration Test
+
+on:
+  schedule:
+    # Run weekdays (Monday through Thursday) at 02:00 UTC (tests latest images)
+    - cron: '0 2 * * 1-4'
+
+  release:
+    types: [published]
+  workflow_dispatch:
+    # Allow manual trigger via GitHub UI with optional tag override
+    inputs:
+      version_tag:
+        description: 'Image tag to test (e.g. latest, stable, 1.1.0). If empty, defaults to latest.'
+        required: false
+        type: string
+        default: 'latest'
+      confirm_resource_cost:
+        description: 'Type "yes" to authorize the high GCP sandbox resource cost of this run.'
+        required: true
+        type: string
+        default: 'no'
+
+permissions:
+  contents: 'read'
+  id-token: 'write'
+
+jobs:
+  gcp-integration-test:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Verify Confirmation (Manual Runs Only)
+        if: github.event_name == 'workflow_dispatch'
+        run: |
+          CONFIRMATION="${{ github.event.inputs.confirm_resource_cost }}"
+          if [ "${CONFIRMATION,,}" != "yes" ]; then
+            echo "Error: High-cost execution confirmation was not authorized (must type 'yes')."
+            exit 1
+          fi
+
+      - name: Checkout repository
+        uses: actions/checkout@v5
+
+      - name: Authenticate to GCP
+        uses: google-github-actions/auth@v2
+        with:
+          workload_identity_provider: ${{ secrets.GCP_WIF_PROVIDER }}
+          service_account: ${{ secrets.GCP_WIF_SA_EMAIL }}
+
+      - name: Set up GCP Cloud SDK
+        uses: google-github-actions/setup-gcloud@v2
+        with:
+          project_id: datcom-ci
+
+      - name: Install uv and Set up Python
+        uses: astral-sh/setup-uv@v6
+        with:
+          python-version: "3.12"
+
+      - name: Install Dependencies
+        run: uv sync --locked
+
+      - name: Determine Image Tag
+        id: image-tag
+        run: |
+          if [ "${{ github.event_name }}" = "release" ]; then
+            # Strip 'v' prefix if present (e.g. v1.1.1 -> 1.1.1)
+            RAW_TAG="${{ github.event.release.tag_name }}"
+            TAG="${RAW_TAG#v}"
+            echo "Using release image tag: $TAG"
+            echo "tag=$TAG" >> $GITHUB_OUTPUT
+          elif [ "${{ github.event_name }}" = "workflow_dispatch" ]; then
+            TAG="${{ github.event.inputs.version_tag }}"
+            echo "Using manual input image tag: $TAG"
+            echo "tag=$TAG" >> $GITHUB_OUTPUT
+          else
+            echo "Using default cron image tag: latest"
+            echo "tag=latest" >> $GITHUB_OUTPUT
+          fi
+
+      - name: Run GCP Integration Test
+        env:
+          DC_API_KEY: ${{ secrets.DC_API_KEY }}
+        run: |
+          TAG="${{ steps.image-tag.outputs.tag }}"
+          
+          uv run python3 tests/datacommons-integration-tests/gcp_test_runner.py \
+            --project-id datcom-ci \
+            --dcp-version "$TAG" \
+            --dc-api-key "$DC_API_KEY"
+
+
diff --git a/tests/datacommons-integration-tests/README.md b/tests/datacommons-integration-tests/README.md
index 9a7e0b85..cb95bc04 100644
--- a/tests/datacommons-integration-tests/README.md
+++ b/tests/datacommons-integration-tests/README.md
@@ -60,30 +60,51 @@ You can modify the flags inside `custom_feature_flags.yaml` if you need to tweak
 
 In addition to local emulated tests, this package includes a script to run automated, end-to-end integration tests on GCP real resources. This validates Terraform scaffolding, GCS, Spanner database seeding, Cloud Workflow executions, and Cloud Run web server APIs.
 
-### Running the GCP Tests
+### Execution Modes
+
+The GCP integration runner supports two execution modes:
+
+#### 1. Heavy Mode (Full Provisioning & E2E Validation)
+Pulls the latest templates, provisions a completely new isolated sandbox stack in GCP via Terraform (Spanner DB, GCS bucket, Cloud Workflows, and Cloud Run service), performs ingestion, runs the tests, and destroys the sandbox on completion (unless `--keep-sandbox` is specified).
+* **Usage:**
+  ```bash
+  uv run python tests/datacommons-integration-tests/gcp_test_runner.py \
+      --project-id datcom-ci \
+      --dc-api-key "YOUR_API_KEY"
+  ```
+
+#### 2. Lightweight Mode (Fast Debugging Loop)
+Skips the slow Terraform setup and teardown stages entirely. It reads resource configurations from an existing `/tmp/workspace-<namespace>` directory, clears the database tables, re-uploads local test data files, triggers the ingestion workflow, and runs `pytest`. This reduces iteration time **from 10+ minutes down to 1-2 minutes**.
+* **Usage:**
+  1. First, create a persistent sandbox and keep it alive:
+     ```bash
+     uv run python tests/datacommons-integration-tests/gcp_test_runner.py --namespace my-debug-sandbox --keep-sandbox
+     ```
+  2. Run subsequent test runs in lightweight mode:
+     ```bash
+     uv run python tests/datacommons-integration-tests/gcp_test_runner.py --namespace my-debug-sandbox --reuse-sandbox
+     ```
 
-To run the GCP integration test suite (requires authenticated GCP CLI access):
-```bash
-uv run python tests/datacommons-integration-tests/run_gcp_integration_test.py \
-    --project-id <YOUR_GCP_PROJECT_ID> \
-    --dc-api-key "YOUR_GOOGLE_DATA_COMMONS_API_KEY"
-```
-
-Alternatively, you can pass the API key via an environment variable:
-```bash
-DC_API_KEY="YOUR_API_KEY" uv run python tests/datacommons-integration-tests/run_gcp_integration_test.py \
-    --project-id <YOUR_GCP_PROJECT_ID>
-```
 
 ### Script Arguments
-*   `--project-id`: GCP Project ID (default: `datcom-ci`).
-*   `--dc-api-key`: Data Commons API Key.
-*   `--region`: GCP region (default: `us-central1`).
-*   `--namespace`: Custom resource naming namespace (default: randomized `itest-XXXX`).
-*   `--keep-sandbox`: Do not destroy sandbox GCP resources on completion (useful for debugging).
-*   `--reuse-sandbox`: Reuse existing sandbox resources (requires passing persistent `--namespace` and having run with `--keep-sandbox` previously).
-*   `--tf-git-ref`: Git reference for the Terraform templates repository (default: `main`).
-*   `--services-image`, `--preprocessing-image`, `--helper-image`: Override container images deployed during provisioning.
+*   `--project-id`: The Google Cloud Project ID where the sandbox resources should be provisioned (default: `datcom-ci`).
+*   `--dc-api-key`: Google Data Commons API Key needed to authenticate and configure sandbox clients.
+*   `--region`: The GCP region to deploy resources (default: `us-central1`).
+*   `--namespace`: Custom naming namespace for resources (default: randomized `itest-XXXX`).
+*   `--keep-sandbox`: Do not destroy sandbox GCP resources on completion/failure. Useful for debugging active instances.
+*   `--reuse-sandbox`: Reuse existing local workspace and GCP sandbox resources if they exist. Requires passing a persistent, custom `--namespace` (e.g. `--namespace itest-9611`) and having run with `--keep-sandbox` in the previous run.
+*   `--tf-git-ref`: Git reference branch/tag/commit for the GCP Terraform templates repository (default: `main`).
+*   `--dcp-version`: Override default DCP version (controls all images and templates) (default: `latest`).
+
+
+### E2E Verification Stages & Philosophy
+
+Once the GCP sandbox stack is up and seeded with dataset MCFs, the script triggers the entire API verification suite concurrently. This validates three core capabilities of the deployed Data Commons Platform backend:
+
+*   **Stage A: V2 Observations & Custom Data Retrieval**
+    *   *Philosophy:* Validates E2E query routing and data loading for custom variables. It asserts observations values on custom seeded OECD wages metrics to verify local database mapping, loading, and seeding.
+*   **Stage B: V2 Embeddings & Natural Language Resolution**
+    *   *Philosophy:* Validates custom search index and embeddings generation. This checks that the Ingestion Helper successfully generated vector coordinates from custom MCFs, loaded them into Spanner, and that the Resolver can map natural language queries (like "wages") back to these custom database entities.
 
 
 
diff --git a/tests/datacommons-integration-tests/run_gcp_integration_test.py b/tests/datacommons-integration-tests/gcp_test_runner.py
similarity index 72%
rename from tests/datacommons-integration-tests/run_gcp_integration_test.py
rename to tests/datacommons-integration-tests/gcp_test_runner.py
index 0d90dc5d..fb3676dd 100755
--- a/tests/datacommons-integration-tests/run_gcp_integration_test.py
+++ b/tests/datacommons-integration-tests/gcp_test_runner.py
@@ -21,7 +21,6 @@
 """
 
 import argparse
-import concurrent.futures
 import json
 import logging
 import os
@@ -280,13 +279,8 @@ def configure_tfvars(sandbox_dir: Path, overrides: dict[str, str | int | bool])
     log_success("Added tfvars overrides")
 
 
-def run_terraform_apply(sandbox_dir: Path) -> dict[str, str]:
-    """Runs terraform init, apply, and extracts parsed outputs."""
-    log_step("Provisioning infrastructure via Terraform (this can take 5-7 minutes)...")
-    run_command(["terraform", "init"], cwd=sandbox_dir, retries=3)
-    run_command(["terraform", "apply", "-auto-approve"], cwd=sandbox_dir, retries=3)
-    log_success("Terraform provisioning completed successfully")
-
+def get_terraform_outputs(sandbox_dir: Path) -> dict[str, str]:
+    """Retrieves and parses current Terraform outputs."""
     log_step("Retrieving Terraform outputs...")
     tf_out_proc = run_command(["terraform", "output", "-json"], cwd=sandbox_dir)
     tf_outputs = json.loads(tf_out_proc.stdout)
@@ -303,6 +297,15 @@ def run_terraform_apply(sandbox_dir: Path) -> dict[str, str]:
     return outputs
 
 
+def run_terraform_apply(sandbox_dir: Path) -> dict[str, str]:
+    """Runs terraform init, apply, and extracts parsed outputs."""
+    log_step("Provisioning infrastructure via Terraform (this can take 5-7 minutes)...")
+    run_command(["terraform", "init"], cwd=sandbox_dir, retries=3)
+    run_command(["terraform", "apply", "-auto-approve"], cwd=sandbox_dir, retries=3)
+    log_success("Terraform provisioning completed successfully")
+    return get_terraform_outputs(sandbox_dir)
+
+
 def upload_dataset_to_gcs(bucket_name: str, wages_data_dir: Path) -> None:
     """Verifies test files and uploads observations, schema, and configs to GCS."""
     log_step(f"Uploading files to GCS bucket: gs://{bucket_name}/ingestion/input/...")
@@ -514,122 +517,7 @@ def verify_spanner_records(
         )
 
 
-def check_api_endpoint(
-    test_name: str,
-    test_port: int,
-    path: str,
-    headers: dict[str, str] | None = None,
-    verify_json_fn=None,
-) -> None:
-    """Helper function to run an HTTP API request and verify status_code and JSON response."""
-    url = f"http://127.0.0.1:{test_port}{path}"
-    logging.info(f"Running {test_name}: {url}")
-    req = urllib.request.Request(url)
-    if headers:
-        for k, v in headers.items():
-            req.add_header(k, v)
-    with urllib.request.urlopen(req, timeout=15) as response:
-        status_code = response.getcode()
-        if status_code != 200:
-            raise RuntimeError(f"{test_name} failed with status code {status_code}")
-        resp_body = response.read().decode("utf-8")
-        log_success(f"{test_name} passed (HTTP 200)")
-        if verify_json_fn:
-            resp_json = json.loads(resp_body)
-            verify_json_fn(resp_json)
-
-
-# Helper verifiers for wages data responses.
-
-
-def verify_wages_obs(resp: dict) -> None:
-    """Verifies observation response values for average_annual_wage."""
-    by_var = resp.get("byVariable", {})
-    if "average_annual_wage" not in by_var:
-        raise ValueError(
-            f"Expected variable 'average_annual_wage' in response, got: {resp}"
-        )
-    by_entity = by_var["average_annual_wage"].get("byEntity", {})
-    if "country/USA" not in by_entity or "country/CAN" not in by_entity:
-        raise ValueError(
-            f"Expected entities 'country/USA' and 'country/CAN', got: {by_entity.keys()}"
-        )
-    for entity_dcid, entity_data in by_entity.items():
-        facets = entity_data.get("orderedFacets", [])
-        if not facets:
-            raise ValueError(
-                f"Expected orderedFacets list for {entity_dcid}, got: {entity_data}"
-            )
-        observations = facets[0].get("observations", [])
-        if not observations:
-            raise ValueError(
-                f"Expected observations list for {entity_dcid}, got: {entity_data}"
-            )
-
-
-def verify_wages_resolve(resp: dict) -> None:
-    """Verifies indicator resolution response candidates match expected wages variables."""
-    entities = resp.get("entities", [])
-    if not entities:
-        raise ValueError(f"Expected non-empty 'entities' list, got: {resp}")
-    node_res = entities[0]
-    if node_res.get("node") != "wages":
-        raise ValueError(f"Expected resolved node for 'wages', got: {node_res}")
-    candidates = node_res.get("candidates", [])
-    if not candidates:
-        raise ValueError(f"Expected resolved candidates for 'wages', got: {node_res}")
-    has_sv = any(
-        c.get("dcid") in ("average_annual_wage", "gender_wage_gap") for c in candidates
-    )
-    if not has_sv:
-        raise ValueError(
-            f"Expected resolved candidate to match 'average_annual_wage' or 'gender_wage_gap', got: {candidates}"
-        )
-
-
-API_VERIFICATION_STAGES = {
-    "Stage A: V2 Observations API": [
-        {
-            "name": "Test 1: V2 Multi-entity Observation (SDG)",
-            "path": "/core/api/v2/observation?select=variable&select=entity&select=date&select=value&variable.dcids=undata/sdg/IT_MOB_OWN.SEX--_T&entity.dcids=country/IND&entity.dcids=country/USA",
-            "headers": {"X-Use-Multi-Entity-Schema": "true"},
-        },
-        {
-            "name": "Test 2: Verify Wages Data (Custom Data)",
-            "path": "/core/api/v2/observation?select=variable&select=entity&select=date&select=value&variable.dcids=average_annual_wage&entity.dcids=country/USA&entity.dcids=country/CAN",
-            "headers": {"X-Use-Multi-Entity-Schema": "true"},
-            "verify_json_fn": verify_wages_obs,
-        },
-    ],
-    "Stage B: V2 Bulk Group Info API": [
-        {
-            "name": "Test 3a: V2 Bulk Variable Group Info (Unconstrained)",
-            "path": "/core/api/v2/bulk/info/variable-group?nodes=dc/g/undata/SDG_1.5.1",
-            "headers": {"X-Use-Multi-Entity-Schema": "true"},
-        },
-        {
-            "name": "Test 3b: V2 Bulk Variable Group Info (Constrained by USA)",
-            "path": "/core/api/v2/bulk/info/variable-group?nodes=dc/g/undata/SDG_1.5.1&constrained_entities=country/USA",
-            "headers": {"X-Use-Multi-Entity-Schema": "true"},
-        },
-        {
-            "name": "Test 3c: V2 Bulk Variable Group Info (Constrained by VAT)",
-            "path": "/core/api/v2/bulk/info/variable-group?nodes=dc/g/undata/SDG_1.5.1&constrained_entities=country/VAT",
-            "headers": {"X-Use-Multi-Entity-Schema": "true"},
-        },
-    ],
-    "Stage C: V2 Embeddings & Resolve API": [
-        {
-            "name": "Test 4: V2 Resolve - Indicator resolver (Food Waste)",
-            "path": "/core/api/v2/resolve?nodes=food%20waste&resolver=indicator&target=custom_only",
-        },
-        {
-            "name": "Test 5: V2 Resolve - Indicator resolver (Wages)",
-            "path": "/core/api/v2/resolve?nodes=wages&resolver=indicator&target=custom_only",
-            "verify_json_fn": verify_wages_resolve,
-        },
-    ],
-}
+# Note: API verifications and assertions have been moved to pytest suites in tests/datacommons-integration-tests/suites/
 
 
 def run_serving_proxy_tests(service_name: str, region: str, project_id: str) -> None:
@@ -703,61 +591,18 @@ def run_serving_proxy_tests(service_name: str, region: str, project_id: str) ->
                 f"Failed to fetch a warm container instance containing the typeOf StatisticalVariable mapping after {max_retries * retry_interval}s."
             )
 
-        # Run API verification tests in parallel.
-        all_tests = []
-        for stage_name, tests in API_VERIFICATION_STAGES.items():
-            for t in tests:
-                t_copy = t.copy()
-                t_copy["stage"] = stage_name
-                all_tests.append(t_copy)
-
-        def run_single_test(t: dict) -> tuple[str, str, bool, str | None]:
-            try:
-                check_api_endpoint(
-                    test_name=t["name"],
-                    test_port=test_port,
-                    path=t["path"],
-                    headers=t.get("headers"),
-                    verify_json_fn=t.get("verify_json_fn"),
-                )
-                return (t["name"], t["stage"], True, None)
-            except Exception as test_err:
-                logging.error(f"[FAILURE] {t['name']} failed: {test_err}")
-                return (t["name"], t["stage"], False, str(test_err))
-
-        with concurrent.futures.ThreadPoolExecutor(
-            max_workers=len(all_tests)
-        ) as executor:
-            futures = {executor.submit(run_single_test, t): t for t in all_tests}
-            results = [
-                future.result() for future in concurrent.futures.as_completed(futures)
-            ]
-
-        # Group and print results by stage.
-        by_stage: dict[str, list[tuple[str, bool, str | None]]] = {}
-        for name, stage, success, err in results:
-            by_stage.setdefault(stage, []).append((name, success, err))
-
-        print("\n" + "=" * 65)
-        print("                 API VERIFICATION SUMMARY REPORT")
-        print("=" * 65)
-        failed = False
-        for stage, stage_results in sorted(by_stage.items()):
-            print(f"\n{stage}:")
-            for name, success, err in sorted(stage_results):
-                status = "[PASSED]" if success else "[FAILED]"
-                print(f"  {status:<8} {name}")
-                if not success:
-                    print(f"           Error: {err}")
-                    failed = True
-        print("\n" + "=" * 65)
-
-        if failed:
-            raise RuntimeError(
-                "One or more E2E serving verification tests failed. See report details above."
-            )
-
-        log_success("All parallel verification tests completed successfully!")
+        # Run pytest E2E suites against the proxy port
+        pytest_cmd = [
+            "uv",
+            "run",
+            "pytest",
+            "tests/datacommons-integration-tests/suites/",
+            "--target-url",
+            f"http://127.0.0.1:{test_port}",
+        ]
+        log_step(f"Running pytest suites: {' '.join(pytest_cmd)}")
+        run_command(pytest_cmd, check=True)
+        log_success("All pytest verification suites passed successfully!")
 
     except Exception as e:
         console_url = f"https://console.cloud.google.com/run/detail/{region}/{service_name}/logs?project={project_id}"
@@ -864,26 +709,21 @@ def main() -> None:
         help="Custom namespace (defaults to itest-XXXX)",
     )
     parser.add_argument(
-        "--services-image",
-        default="gcr.io/datcom-ci/datacommons-services:latest",
-        help="Override default serving services container image",
-    )
-    parser.add_argument(
-        "--preprocessing-image",
-        default="gcr.io/datcom-ci/datacommons-data:latest",
-        help="Override default data preprocessing image",
-    )
-    # Use stable helper tag by default until stable/local pushes are fixed for ingestion (aligning with the default in tests/datacommons-integration-tests/docker-compose.test.yml).
-    parser.add_argument(
-        "--helper-image",
-        default="gcr.io/datcom-ci/datacommons-ingestion-helper:stable",
-        help="Override default helper service image",
+        "--dcp-version",
+        default="latest",
+        help="Override default DCP version (controls all images and templates)",
     )
+
     parser.add_argument(
         "--keep-sandbox",
         action="store_true",
         help="Do not destroy sandbox on completion/failure",
     )
+    parser.add_argument(
+        "--reuse-sandbox",
+        action="store_true",
+        help="Skip terraform provision/destroy steps and reuse existing sandbox workspace",
+    )
     parser.add_argument(
         "--dc-api-key",
         default=os.environ.get("DC_API_KEY", ""),
@@ -918,56 +758,61 @@ def main() -> None:
     sandbox_dir = None
 
     try:
-        # 1. Setup workspace
-        setup_workspace(workspace_dir, namespace)
-
-        # 1b. Validate local MCF schema syntax before deploying expensive cloud resources
-        root_dir = Path(__file__).resolve().parent.parent.parent
-        wages_data_dir = root_dir / "samples" / "OECD_wage_data"
-        validate_mcf_file(wages_data_dir / "average_annual_wage.mcf")
-        validate_mcf_file(wages_data_dir / "gender_wage_gap.mcf")
+        if args.reuse_sandbox:
+            if not args.namespace:
+                raise RuntimeError(
+                    "Namespace must be specified when using --reuse-sandbox (e.g. --namespace itest-XXXX)"
+                )
+            sandbox_dir = workspace_dir / namespace
+            if not sandbox_dir.exists():
+                raise RuntimeError(f"Sandbox directory does not exist: {sandbox_dir}")
+            log_step(f"Reusing existing sandbox namespace: {namespace}")
+            outputs = get_terraform_outputs(sandbox_dir)
+        else:
+            # 1. Setup workspace
+            setup_workspace(workspace_dir, namespace)
+
+            # 1b. Validate local MCF schema syntax before deploying expensive cloud resources
+            root_dir = Path(__file__).resolve().parent.parent.parent
+            wages_data_dir = root_dir / "samples" / "OECD_wage_data"
+            validate_mcf_file(wages_data_dir / "average_annual_wage.mcf")
+            validate_mcf_file(wages_data_dir / "gender_wage_gap.mcf")
+
+            # 2. Initialize configuration via CLI and setup overrides
+            dc_key = resolve_api_key(args.project_id, args.dc_api_key)
+            if not dc_key:
+                raise RuntimeError(
+                    "Data Commons API Key must be provided (either via --dc-api-key flag, "
+                    "DC_API_KEY environment variable, or 'dc-api-key' Secret in GCP Secret Manager)."
+                )
 
-        # 2. Initialize configuration via CLI and setup overrides
-        dc_key = resolve_api_key(args.project_id, args.dc_api_key)
-        if not dc_key:
-            raise RuntimeError(
-                "Data Commons API Key must be provided (either via --dc-api-key flag, "
-                "DC_API_KEY environment variable, or 'dc-api-key' Secret in GCP Secret Manager)."
+            sandbox_dir = initialize_sandbox_scaffolding(
+                workspace_root=workspace_root,
+                workspace_dir=workspace_dir,
+                project_id=args.project_id,
+                namespace=namespace,
+                dc_api_key=dc_key,
+                tf_git_ref=args.tf_git_ref,
+                tf_state_bucket=args.tf_state_bucket,
             )
 
-        sandbox_dir = initialize_sandbox_scaffolding(
-            workspace_root=workspace_root,
-            workspace_dir=workspace_dir,
-            project_id=args.project_id,
-            namespace=namespace,
-            dc_api_key=dc_key,
-            tf_git_ref=args.tf_git_ref,
-            tf_state_bucket=args.tf_state_bucket,
-        )
-
-        # 3. Configure tfvars overrides
-        tfvars_overrides: dict[str, str | int | bool] = {
-            "spanner_create_instance": False,
-            "spanner_instance_id": "dcp-integration-test-shared-instance",
-            "spanner_create_database": True,
-            "spanner_create_bigquery_reservation": False,
-            "datacommons_services_min_instances": 1,
-            "datacommons_services_max_instances": 1,
-        }
-        if args.services_image:
-            tfvars_overrides["datacommons_services_image"] = args.services_image
-        if args.preprocessing_image:
-            tfvars_overrides["ingestion_preprocessing_job_image"] = (
-                args.preprocessing_image
-            )
-        if args.helper_image:
-            tfvars_overrides["ingestion_helper_service_image"] = args.helper_image
+            # 3. Configure tfvars overrides
+            tfvars_overrides: dict[str, str | int | bool] = {
+                "spanner_create_instance": False,
+                "spanner_instance_id": "dcp-integration-test-shared-instance",
+                "spanner_create_database": True,
+                "spanner_create_bigquery_reservation": False,
+                "datacommons_services_min_instances": 1,
+                "datacommons_services_max_instances": 1,
+            }
+            if args.dcp_version:
+                tfvars_overrides["dcp_version"] = args.dcp_version
 
-        configure_tfvars(sandbox_dir, tfvars_overrides)
+            configure_tfvars(sandbox_dir, tfvars_overrides)
 
-        # 4. Provision Infrastructure (Terraform Apply)
-        terraform_provisioned = True
-        outputs = run_terraform_apply(sandbox_dir)
+            # 4. Provision Infrastructure (Terraform Apply)
+            terraform_provisioned = True
+            outputs = run_terraform_apply(sandbox_dir)
 
         # 5. Locate and upload OECD Wages Dataset files to GCS
         root_dir = Path(__file__).resolve().parent.parent.parent
@@ -986,11 +831,12 @@ def main() -> None:
             workflow_name=outputs["workflow_name"],
         )
 
-        # 8. Query Spanner to verify data was loaded
+        # 8. Query Spanner to verify data was loaded (Mock datasets add 65 rows)
         verify_spanner_records(
             project_id=args.project_id,
             spanner_database=outputs["spanner_database"],
             spanner_instance=outputs["spanner_instance"],
+            min_expected_rows=60,
         )
 
         # 9. Run Proxy verification to test the serving layers
diff --git a/tests/datacommons-integration-tests/mock_nl_server.py b/tests/datacommons-integration-tests/mock_nl_server.py
index 7c8bb5ca..21626818 100644
--- a/tests/datacommons-integration-tests/mock_nl_server.py
+++ b/tests/datacommons-integration-tests/mock_nl_server.py
@@ -24,8 +24,8 @@
   weights at test runtime which introduces flakiness and violates offline-hermeticity.
 
 Thus, we spin up this lightweight 0-dependency mock server in python to intercept
-NL search requests (like "Number of frogs") and map them to seeded Spanner variables
-(like "number_of_frogs"). This allows us to test the happy path end-to-end (querying
+NL search requests (like "Average annual wage") and map them to seeded Spanner variables
+(like "average_annual_wage"). This allows us to test the happy path end-to-end (querying
 spanner and asserting observation charts) without real model serving infrastructure.
 
 TODO: Remove this mock server and use a proper local SentenceTransformers
diff --git a/tests/datacommons-integration-tests/suites/conftest.py b/tests/datacommons-integration-tests/suites/conftest.py
new file mode 100644
index 00000000..2275e83a
--- /dev/null
+++ b/tests/datacommons-integration-tests/suites/conftest.py
@@ -0,0 +1,66 @@
+# 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.
+
+import json
+import urllib.request
+
+import pytest
+
+
+def pytest_addoption(parser):
+    parser.addoption(
+        "--target-url",
+        action="store",
+        default="http://localhost:8080",
+        help="Target base URL of the Data Commons serving services instance under test",
+    )
+
+
+@pytest.fixture(scope="session")
+def target_url(request):
+    return request.config.getoption("--target-url").rstrip("/")
+
+
+@pytest.fixture(scope="session")
+def api_client():
+    def _make_request(
+        target_url: str, path: str, headers: dict = None, *, is_json: bool = True
+    ):
+        url = f"{target_url}{path}"
+        if not url.startswith(("http://", "https://")):
+            raise ValueError(f"URL scheme must be http or https, got: {url}")
+        req = urllib.request.Request(url)  # noqa: S310
+        if headers:
+            for k, v in headers.items():
+                req.add_header(k, v)
+        try:
+            with urllib.request.urlopen(req, timeout=15) as response:  # noqa: S310
+                status_code = response.getcode()
+                if status_code != 200:
+                    raise AssertionError(
+                        f"Expected HTTP status code 200, got: {status_code}"
+                    )
+                body = response.read().decode("utf-8")
+                if is_json:
+                    try:
+                        return json.loads(body)
+                    except json.JSONDecodeError as jde:
+                        raise AssertionError(
+                            f"Response body is not valid JSON: {jde}. Raw body snippet (first 500 chars): {body[:500]}"
+                        ) from jde
+                return body
+        except Exception as e:  # noqa: BLE001
+            raise AssertionError(f"HTTP request to {url} failed: {e}") from e
+
+    return _make_request
diff --git a/tests/datacommons-integration-tests/suites/test_oecd_wages.py b/tests/datacommons-integration-tests/suites/test_oecd_wages.py
new file mode 100644
index 00000000..e145317f
--- /dev/null
+++ b/tests/datacommons-integration-tests/suites/test_oecd_wages.py
@@ -0,0 +1,99 @@
+# 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.
+
+
+def test_v2_multi_entity_observation_sdg(target_url, api_client):
+    """Test 1: V2 Multi-entity Observation (SDG)"""
+    path = "/core/api/v2/observation?select=variable&select=entity&select=date&select=value&variable.dcids=undata/sdg/IT_MOB_OWN.SEX--_T&entity.dcids=country/IND&entity.dcids=country/USA"
+    # Just verify it returns 200 OK
+    api_client(target_url, path, headers={"X-Use-Multi-Entity-Schema": "true"})
+
+
+def test_verify_wages_data(target_url, api_client):
+    """Test 2: Verify Wages Data (Custom Data)"""
+    path = "/core/api/v2/observation?select=variable&select=entity&select=date&select=value&variable.dcids=average_annual_wage&entity.dcids=country/USA&entity.dcids=country/CAN"
+    resp = api_client(target_url, path, headers={"X-Use-Multi-Entity-Schema": "true"})
+
+    by_var = resp.get("byVariable", {})
+    assert "average_annual_wage" in by_var, (
+        f"Expected variable 'average_annual_wage' in response, got: {resp}"
+    )
+
+    by_entity = by_var["average_annual_wage"].get("byEntity", {})
+    assert "country/USA" in by_entity, (
+        f"Expected entity 'country/USA' in response, got: {by_entity.keys()}"
+    )
+    assert "country/CAN" in by_entity, (
+        f"Expected entity 'country/CAN' in response, got: {by_entity.keys()}"
+    )
+
+    for entity_dcid, entity_data in by_entity.items():
+        facets = entity_data.get("orderedFacets", [])
+        assert facets, (
+            f"Expected orderedFacets list for {entity_dcid}, got: {entity_data}"
+        )
+        observations = facets[0].get("observations", [])
+        assert observations, (
+            f"Expected observations list for {entity_dcid}, got: {entity_data}"
+        )
+
+
+def test_v2_bulk_variable_group_info_unconstrained(target_url, api_client):
+    """Test 3a: V2 Bulk Variable Group Info (Unconstrained)"""
+    path = "/core/api/v2/bulk/info/variable-group?nodes=dc/g/undata/SDG_1.5.1"
+    api_client(target_url, path, headers={"X-Use-Multi-Entity-Schema": "true"})
+
+
+def test_v2_bulk_variable_group_info_constrained_usa(target_url, api_client):
+    """Test 3b: V2 Bulk Variable Group Info (Constrained by USA)"""
+    path = "/core/api/v2/bulk/info/variable-group?nodes=dc/g/undata/SDG_1.5.1&constrained_entities=country/USA"
+    api_client(target_url, path, headers={"X-Use-Multi-Entity-Schema": "true"})
+
+
+def test_v2_bulk_variable_group_info_constrained_vat(target_url, api_client):
+    """Test 3c: V2 Bulk Variable Group Info (Constrained by VAT)"""
+    path = "/core/api/v2/bulk/info/variable-group?nodes=dc/g/undata/SDG_1.5.1&constrained_entities=country/VAT"
+    api_client(target_url, path, headers={"X-Use-Multi-Entity-Schema": "true"})
+
+
+def test_v2_resolve_indicator_food_waste(target_url, api_client):
+    """Test 4: V2 Resolve - Indicator resolver (Food Waste)"""
+    path = (
+        "/core/api/v2/resolve?nodes=food%20waste&resolver=indicator&target=custom_only"
+    )
+    api_client(target_url, path)
+
+
+def test_v2_resolve_indicator_wages(target_url, api_client):
+    """Test 5: V2 Resolve - Indicator resolver (Wages)"""
+    path = "/core/api/v2/resolve?nodes=wages&resolver=indicator&target=custom_only"
+    resp = api_client(target_url, path)
+
+    entities = resp.get("entities", [])
+    assert entities, f"Expected non-empty 'entities' list, got: {resp}"
+
+    node_res = entities[0]
+    assert node_res.get("node") == "wages", (
+        f"Expected resolved node for 'wages', got: {node_res}"
+    )
+
+    candidates = node_res.get("candidates", [])
+    assert candidates, f"Expected resolved candidates for 'wages', got: {node_res}"
+
+    has_sv = any(
+        c.get("dcid") in ("average_annual_wage", "gender_wage_gap") for c in candidates
+    )
+    assert has_sv, (
+        f"Expected resolved candidate to match 'average_annual_wage' or 'gender_wage_gap', got: {candidates}"
+    )