chore: merge latest main

Signed-off-by: jeremyhi <fengjiachun@gmail.com>
2026-10-26 05:36:16 +00:00 · 2026-08-31 13:01:27 +08:00
parent 5af1aa0104 d32cd77505
commit 865a29584b
622 changed files with 71033 additions and 11579 deletions
@@ -0,0 +1,485 @@
+#!/usr/bin/env python3
+# Copyright 2023 Greptime Team
+#
+# 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.
+
+# This is the first `.github/scripts/` user of PEP 723 inline script metadata:
+# `uv run` installs the pinned Aliyun SDK before executing. The SDK import stays
+# lazy (inside `make_ecs_client`) so unit tests can import this module with a
+# plain stdlib interpreter.
+# /// script
+# requires-python = ">=3.10"
+# dependencies = [
+#   # Pin exactly after the first live run against the Aliyun API.
+#   "alibabacloud_ecs20140526>=4.1.0,<6",
+#   "alibabacloud_tea_openapi>=0.3.12,<1",
+# ]
+# ///
+
+"""Provision an ephemeral Aliyun ECS query-regression runner.
+
+Creates one pay-as-you-go ECS instance from the query-regression custom image
+and waits until the instance registers itself as an ephemeral GitHub Actions
+runner with a per-run label. Build caches live on the instance system disk
+and disappear with the VM. The instance id is written to job outputs as soon
+as the VM exists so the teardown job is the single DeleteInstance caller (a
+timeout here used to delete and then teardown deleted again). On runner
+online timeout the console is dumped and the job fails; teardown releases
+the instance.
+
+The Aliyun credentials are used only in this control-plane job (ubuntu-latest)
+and are never passed to the ECS instance; the instance receives only a
+short-lived GitHub runner registration token via user data.
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import os
+import time
+import urllib.error
+import urllib.request
+from dataclasses import dataclass
+
+RUNNER_NAME_PREFIX = "qreg-ecs"
+RUNNER_LABEL_PREFIX = "query-regression-ecs"
+MANAGED_BY_TAG_KEY = "managed-by"
+MANAGED_BY_TAG_VALUE = "query-regression-ci"
+RUN_TAG_KEY = "query-regression-run-id"
+
+# Runner cache paths on the instance system disk. They are created empty
+# every provision and deleted with the VM.
+CACHE_PATHS = (
+    "/home/runner/.cargo/registry",
+    "/home/runner/.cargo/git",
+    "/home/runner/query-regression-target",
+    "/home/runner/query-regression-cache-meta",
+    "/home/runner/.cache/sccache",
+)
+
+RUNNER_ONLINE_TIMEOUT_SECONDS = 10 * 60
+POLL_INTERVAL_SECONDS = 5
+
+# Nightly thin-LTO of greptime peaks above ecs.c9i.2xlarge's 16 GiB. A swap
+# file plus masking systemd-oomd lets the kernel reclaim rustc pages instead
+# of SIGTERM-ing the runner service cgroup (GitHub then reports the step as
+# "The operation was canceled"). Sized to match that 16 GiB instance; a
+# 32 GiB type still benefits as a safety net. Lives on the system disk next
+# to the cold build caches.
+SWAP_FILE = "/swapfile"
+SWAP_SIZE_GIB = 16
+
+
+@dataclass(frozen=True)
+class ProvisionConfig:
+    region_id: str
+    vswitch_id: str
+    security_group_id: str
+    image_id: str
+    instance_type: str
+    repo: str
+    run_id: str
+    github_token: str
+    # Optional resource group; required when the RAM grant is scoped to one.
+    resource_group_id: str | None = None
+    # Runner identity inside the image; the workflow asserts the same values.
+    runner_uid: str = "1001"
+    runner_gid: str = "1001"
+
+
+def runner_name_for_run(run_id: str) -> str:
+    return f"{RUNNER_NAME_PREFIX}-{run_id}"
+
+
+def runner_label_for_run(run_id: str) -> str:
+    return f"{RUNNER_LABEL_PREFIX}-{run_id}"
+
+
+def render_user_data(
+    runner_name: str,
+    runner_label: str,
+    runner_token: str,
+    repo: str,
+    runner_uid: str = "1001",
+    runner_gid: str = "1001",
+) -> str:
+    """Render the cloud-init shell script for the runner instance."""
+    destinations = " ".join(f'"{dst}"' for dst in CACHE_PATHS)
+    cache_setup = f"""# Caches live on the system disk, are deleted with the instance, and every
+# run compiles cold. Within-run reuse (base warming candidate via the shared
+# target dir and sccache) still applies.
+mkdir -p {destinations}
+chown -R {runner_uid}:{runner_gid} /home/runner"""
+    swap_setup = f"""# Mask systemd-oomd before enabling swap: Ubuntu 24.04 kills the whole
+# service cgroup on PSI pressure, and swap thrashing looks like pressure.
+systemctl disable --now systemd-oomd.socket systemd-oomd.service || true
+systemctl mask systemd-oomd.socket systemd-oomd.service || true
+# apt-daily-upgrade / unattended-upgrades can `systemctl restart` services
+# whose libraries were updated. That SIGTERM-s the runner mid-job; GitHub
+# records UserCancelled (job still valid, child exit 143). These VMs live
+# ~1h; security updates belong in the image, not at job time.
+systemctl disable --now unattended-upgrades.service apt-daily.timer apt-daily-upgrade.timer apt-daily.service apt-daily-upgrade.service || true
+systemctl mask unattended-upgrades.service apt-daily.timer apt-daily-upgrade.timer apt-daily.service apt-daily-upgrade.service || true
+systemctl stop unattended-upgrades.service || true
+killall -9 unattended-upgr || true
+cat > /etc/apt/apt.conf.d/99disable-auto-updates <<'APTEOF'
+APT::Periodic::Update-Package-Lists "0";
+APT::Periodic::Unattended-Upgrade "0";
+APT::Periodic::Download-Upgradeable-Packages "0";
+APTEOF
+if [[ ! -f "{SWAP_FILE}" ]]; then
+  fallocate --length {SWAP_SIZE_GIB}G "{SWAP_FILE}"
+  chmod 600 "{SWAP_FILE}"
+  mkswap "{SWAP_FILE}"
+fi
+swapon "{SWAP_FILE}"
+sysctl --write vm.swappiness=10
+swapon --show
+free --human"""
+    return f"""#!/bin/bash
+set -euo pipefail
+
+{cache_setup}
+
+{swap_setup}
+
+cat > /etc/ephemeral-github-runner.env <<'ENVEOF'
+RUNNER_NAME={runner_name}
+RUNNER_LABELS={runner_label}
+RUNNER_TOKEN={runner_token}
+REPO_URL=https://github.com/{repo}
+# The container image baked /opt/cargo/bin into PATH via Dockerfile ENV; on
+# the host nothing inherits that, so publish it through the unit's
+# EnvironmentFile: runner job processes inherit the runner's environment.
+PATH=/opt/cargo/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
+ENVEOF
+
+# Stream the runner service's own logs (config output, job lifecycle, crash
+# messages) to the serial console, next to the kernel's OOM-killer records.
+# Teardown dumps the console tail before deleting the instance, so the
+# witness of what killed the runner survives the machine.
+mkdir -p /etc/systemd/system/ephemeral-github-runner.service.d
+cat > /etc/systemd/system/ephemeral-github-runner.service.d/console.conf <<'CONFEOF'
+[Service]
+StandardOutput=journal+console
+StandardError=journal+console
+CONFEOF
+# systemd's DefaultOOMPolicy=stop SIGTERM-s the whole unit when *any*
+# process in the cgroup is OOM-killed (typically rustc during thin LTO).
+# That is the "Session terminated, killing shell..." / job Canceled path:
+# runuser dies, GitHub deletes the ephemeral registration, always() steps
+# never run. continue keeps the runner up so cargo can report signal 9.
+cat > /etc/systemd/system/ephemeral-github-runner.service.d/oom.conf <<'CONFEOF'
+[Service]
+OOMPolicy=continue
+CONFEOF
+systemctl daemon-reload
+# restart (not start): the image enables this unit, so it may already
+# have come up without the drop-ins if it raced cloud-init. restart
+# applies OOMPolicy=continue to the running cgroup. --no-block matters:
+# newer image units carry After=cloud-final.service, and this script IS
+# cloud-final — a blocking restart would deadlock until TimeoutStartSec.
+systemctl restart --no-block ephemeral-github-runner.service
+"""
+
+
+def encode_user_data(script: str) -> str:
+    return base64.b64encode(script.encode("utf-8")).decode("ascii")
+
+
+def github_api(token: str, method: str, path: str, body: dict | None = None) -> dict:
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    request = urllib.request.Request(
+        f"https://api.github.com{path}",
+        data=data,
+        method=method,
+        headers={
+            "Accept": "application/vnd.github+json",
+            "Authorization": f"Bearer {token}",
+            "X-GitHub-Api-Version": "2022-11-28",
+        },
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=30) as response:
+            return json.loads(response.read().decode("utf-8"))
+    except urllib.error.HTTPError as error:
+        # GitHub's error body says exactly why (e.g. "Must have admin rights to
+        # Repository" for a PAT without the required scope); surface it instead
+        # of a bare "HTTP Error 403".
+        body = error.read().decode("utf-8", "replace")
+        raise SystemExit(
+            f"GitHub API {method} {path} failed: HTTP {error.code}: {body}\n"
+            "The token comes from the GH_PERSONAL_ACCESS_TOKEN secret; it needs "
+            "'repo' scope (classic PAT) or 'Administration: write' on the "
+            "repository (fine-grained PAT)."
+        ) from error
+
+
+def create_registration_token(github_token: str, repo: str) -> str:
+    response = github_api(
+        github_token,
+        "POST",
+        f"/repos/{repo}/actions/runners/registration-token",
+        body={},
+    )
+    return response["token"]
+
+
+def find_runner_by_name(github_token: str, repo: str, name: str) -> dict | None:
+    page = 1
+    while True:
+        response = github_api(
+            github_token,
+            "GET",
+            f"/repos/{repo}/actions/runners?per_page=100&page={page}",
+        )
+        runners = response.get("runners", [])
+        for runner in runners:
+            if runner.get("name") == name:
+                return runner
+        if len(runners) < 100:
+            return None
+        page += 1
+
+
+def make_ecs_client(config: ProvisionConfig):
+    from alibabacloud_ecs20140526.client import Client as EcsClient
+    from alibabacloud_tea_openapi.models import Config as OpenApiConfig
+
+    access_key_id = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID", "")
+    access_key_secret = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET", "")
+    if not access_key_id or not access_key_secret:
+        raise SystemExit(
+            "ALIBABA_CLOUD_ACCESS_KEY_ID/SECRET are empty; in CI they come from the "
+            "ALICLOUD_ECS_ACCESS_KEY_ID/SECRET repository secrets (a missing or "
+            "misnamed secret expands to an empty string)."
+        )
+    return EcsClient(
+        OpenApiConfig(
+            access_key_id=access_key_id,
+            access_key_secret=access_key_secret,
+            region_id=config.region_id,
+            endpoint=f"ecs.{config.region_id}.aliyuncs.com",
+        )
+    )
+
+
+def wait_for_instance_status(client, region_id: str, instance_id: str, wanted: str, deadline: float) -> None:
+    from alibabacloud_ecs20140526 import models as ecs_models
+
+    while time.monotonic() < deadline:
+        response = client.describe_instances(
+            ecs_models.DescribeInstancesRequest(
+                region_id=region_id, instance_ids=json.dumps([instance_id])
+            )
+        )
+        instances = response.body.instances.instance
+        if instances and instances[0].status == wanted:
+            return
+        time.sleep(POLL_INTERVAL_SECONDS)
+    raise TimeoutError(f"Instance {instance_id} did not reach status {wanted} in time")
+
+
+def fetch_console_output(client, region_id: str, instance_id: str) -> str:
+    from alibabacloud_ecs20140526 import models as ecs_models
+
+    response = client.get_instance_console_output(
+        ecs_models.GetInstanceConsoleOutputRequest(region_id=region_id, instance_id=instance_id)
+    )
+    return base64.b64decode(response.body.console_output or "").decode("utf-8", "replace")
+
+
+class ConsoleTailer:
+    """Incrementally prints new serial-console lines so cloud-init progress is
+    visible in the CI log while waiting, not only after a failure dump."""
+
+    def __init__(self, client, region_id: str, instance_id: str) -> None:
+        self.client = client
+        self.region_id = region_id
+        self.instance_id = instance_id
+        self.printed_lines = 0
+
+    def poll(self) -> None:
+        try:
+            lines = fetch_console_output(self.client, self.region_id, self.instance_id).splitlines()
+        except Exception as error:  # noqa: BLE001
+            print(f"[console] unable to fetch console output: {error}", flush=True)
+            return
+        # The API returns a bounded tail; if it ever shrinks, restart from the
+        # beginning of the new buffer rather than skipping lines.
+        if len(lines) < self.printed_lines:
+            self.printed_lines = 0
+        for line in lines[self.printed_lines :]:
+            print(f"[console] {line}", flush=True)
+        self.printed_lines = len(lines)
+
+
+def dump_console_output(client, region_id: str, instance_id: str) -> None:
+    try:
+        output = fetch_console_output(client, region_id, instance_id)
+    except Exception as error:  # noqa: BLE001
+        print(f"Unable to fetch console output for {instance_id}: {error}", flush=True)
+        return
+    tail = "\n".join(output.splitlines()[-80:])
+    print(f"::group::Console output tail for {instance_id}\n{tail}\n::endgroup::", flush=True)
+    summary_path = os.environ.get("GITHUB_STEP_SUMMARY")
+    if summary_path:
+        with open(summary_path, "a", encoding="utf-8") as summary:
+            summary.write(f"<details><summary>ECS console output ({instance_id})</summary>\n\n```\n")
+            summary.write(tail)
+            summary.write("\n```\n</details>\n")
+
+
+def append_github_output(name: str, value: str) -> None:
+    output_path = os.environ.get("GITHUB_OUTPUT")
+    if output_path:
+        with open(output_path, "a", encoding="utf-8") as output:
+            output.write(f"{name}={value}\n")
+
+
+def run_instance(client, config: ProvisionConfig, user_data: str) -> str:
+    from alibabacloud_ecs20140526 import models as ecs_models
+
+    request = ecs_models.RunInstancesRequest(
+        region_id=config.region_id,
+        image_id=config.image_id,
+        resource_group_id=config.resource_group_id,
+        instance_type=config.instance_type,
+        v_switch_id=config.vswitch_id,
+        security_group_id=config.security_group_id,
+        instance_name=runner_name_for_run(config.run_id),
+        host_name=runner_name_for_run(config.run_id),
+        description=f"Ephemeral query-regression runner for run {config.run_id}",
+        amount=1,
+        instance_charge_type="PostPaid",
+        spot_strategy="NoSpot",
+        internet_charge_type="PayByTraffic",
+        internet_max_bandwidth_out=100,
+        # Holds the image (~12G), the 16G swapfile, the source checkout, the
+        # base+candidate cluster data homes, and cold build caches. Deleted
+        # with the instance.
+        system_disk=ecs_models.RunInstancesRequestSystemDisk(category="cloud_essd", size="40"),
+        user_data=user_data,
+        tag=[
+            ecs_models.RunInstancesRequestTag(key=MANAGED_BY_TAG_KEY, value=MANAGED_BY_TAG_VALUE),
+            ecs_models.RunInstancesRequestTag(key=RUN_TAG_KEY, value=config.run_id),
+        ],
+    )
+    response = client.run_instances(request)
+    instance_id = response.body.instance_id_sets.instance_id_set[0]
+    return instance_id
+
+
+def provision(config: ProvisionConfig) -> int:
+    client = make_ecs_client(config)
+    runner_name = runner_name_for_run(config.run_id)
+    runner_label = runner_label_for_run(config.run_id)
+
+    registration_token = create_registration_token(config.github_token, config.repo)
+    user_data = encode_user_data(
+        render_user_data(
+            runner_name,
+            runner_label,
+            registration_token,
+            config.repo,
+            config.runner_uid,
+            config.runner_gid,
+        )
+    )
+
+    print(f"::group::Run instance {runner_name}", flush=True)
+    instance_id = run_instance(client, config, user_data)
+    print(f"Instance id: {instance_id}", flush=True)
+    # Teardown is the only DeleteInstance caller. Write outputs immediately
+    # so a later status/online failure still releases this VM.
+    append_github_output("label", runner_label)
+    append_github_output("instance_id", instance_id)
+    append_github_output("runner_name", runner_name)
+    wait_for_instance_status(client, config.region_id, instance_id, "Running", time.monotonic() + 5 * 60)
+    print("::endgroup::", flush=True)
+
+    print("::group::Wait for runner online", flush=True)
+    deadline = time.monotonic() + RUNNER_ONLINE_TIMEOUT_SECONDS
+    tailer = ConsoleTailer(client, config.region_id, instance_id)
+    last_console_poll = 0.0
+    while time.monotonic() < deadline:
+        runner = find_runner_by_name(config.github_token, config.repo, runner_name)
+        if runner is not None:
+            status = runner.get("status")
+            print(f"Runner {runner_name} status: {status}", flush=True)
+            if status == "online":
+                print("::endgroup::", flush=True)
+                print(f"Runner {runner_name} is online with label {runner_label}", flush=True)
+                return 0
+        # The serial console lags the guest by a minute or so; 30s polling is
+        # enough to follow cloud-init without tripping API throttling.
+        if time.monotonic() - last_console_poll >= 30:
+            tailer.poll()
+            last_console_poll = time.monotonic()
+        time.sleep(POLL_INTERVAL_SECONDS)
+
+    print("::endgroup::", flush=True)
+    print(
+        f"Runner {runner_name} did not come online in time; "
+        "leaving instance for the teardown job to delete",
+        flush=True,
+    )
+    dump_console_output(client, config.region_id, instance_id)
+    return 1
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--region-id", default=os.environ.get("ALIYUN_ECS_REGION_ID"))
+    parser.add_argument("--vswitch-id", default=os.environ.get("ALIYUN_ECS_VSWITCH_ID"))
+    parser.add_argument("--security-group-id", default=os.environ.get("ALIYUN_ECS_SECURITY_GROUP_ID"))
+    parser.add_argument("--image-id", default=os.environ.get("QUERY_REGRESSION_ECS_IMAGE_ID"))
+    parser.add_argument("--instance-type", default=os.environ.get("ALIYUN_ECS_INSTANCE_TYPE"))
+    parser.add_argument("--repo", default=os.environ.get("GITHUB_REPOSITORY"))
+    parser.add_argument("--run-id", default=os.environ.get("GITHUB_RUN_ID"))
+    parser.add_argument("--github-token", default=os.environ.get("GH_PERSONAL_ACCESS_TOKEN"))
+    parser.add_argument("--runner-uid", default=os.environ.get("QUERY_REGRESSION_RUNNER_UID", "1001"))
+    parser.add_argument("--runner-gid", default=os.environ.get("QUERY_REGRESSION_RUNNER_GID", "1001"))
+    parser.add_argument("--resource-group-id", default=os.environ.get("ALIYUN_ECS_RESOURCE_GROUP_ID"))
+    args = parser.parse_args()
+
+    missing = [
+        name
+        for name, value in vars(args).items()
+        if name not in ("resource_group_id",)
+        and (value is None or (isinstance(value, str) and not value))
+    ]
+    if missing:
+        flags = ", ".join(f"--{name.replace('_', '-')}" for name in missing)
+        raise SystemExit(f"Missing required configuration: {flags} (or their env defaults)")
+
+    config = ProvisionConfig(
+        region_id=args.region_id,
+        vswitch_id=args.vswitch_id,
+        security_group_id=args.security_group_id,
+        image_id=args.image_id,
+        instance_type=args.instance_type,
+        repo=args.repo,
+        run_id=args.run_id,
+        github_token=args.github_token,
+        resource_group_id=args.resource_group_id or None,
+        runner_uid=args.runner_uid,
+        runner_gid=args.runner_gid,
+    )
+    return provision(config)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,237 @@
+#!/usr/bin/env python3
+# Copyright 2023 Greptime Team
+#
+# 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.
+
+# PEP 723 inline metadata (see aliyun-ecs-runner-provision.py for the
+# convention). The SDK import stays lazy so unit tests run on a plain
+# stdlib interpreter.
+# /// script
+# requires-python = ">=3.10"
+# dependencies = [
+#   "alibabacloud_ecs20140526>=4.1.0,<6",
+#   "alibabacloud_tea_openapi>=0.3.12,<1",
+# ]
+# ///
+
+"""Tear down ephemeral Aliyun ECS query-regression runners.
+
+Two modes:
+
+- Targeted (per workflow run): delete one instance by id and deregister its
+  runner by name. Both steps are idempotent and best-effort; a missing
+  instance or runner is not an error.
+- Sweep (scheduled janitor): delete every instance tagged as managed by
+  query-regression CI whose creation time is older than the given TTL, and
+  deregister the matching runners. This is the safety net for runs whose
+  teardown job never executed.
+"""
+
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import os
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+# Reuse the GitHub API client, runner lookup, and tag constants from the
+# provision script.
+_PROVISION_SPEC = importlib.util.spec_from_file_location(
+    "aliyun_ecs_runner_provision",
+    Path(__file__).resolve().parent / "aliyun-ecs-runner-provision.py",
+)
+assert _PROVISION_SPEC is not None and _PROVISION_SPEC.loader is not None
+provision = importlib.util.module_from_spec(_PROVISION_SPEC)
+sys.modules[_PROVISION_SPEC.name] = provision
+_PROVISION_SPEC.loader.exec_module(provision)
+
+SWEEP_TTL = timedelta(hours=4)
+# ECS DescribeInstances creation_time is ISO8601 UTC, e.g. "2026-08-17T02:13Z".
+CREATION_TIME_FORMATS = ("%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%dT%H:%MZ")
+
+
+def parse_creation_time(value: str) -> datetime:
+    for fmt in CREATION_TIME_FORMATS:
+        try:
+            return datetime.strptime(value, fmt).replace(tzinfo=timezone.utc)
+        except ValueError:
+            continue
+    raise ValueError(f"Unparsable ECS creation time: {value}")
+
+
+def expired_instance_names(
+    instances: list[tuple[str, str, str]], now: datetime, ttl: timedelta
+) -> list[tuple[str, str]]:
+    """Pick (instance_id, instance_name) pairs whose creation is older than ttl.
+
+    `instances` items are (instance_id, instance_name, creation_time).
+    """
+    expired = []
+    for instance_id, instance_name, creation_time in instances:
+        age = now - parse_creation_time(creation_time)
+        if age >= ttl:
+            expired.append((instance_id, instance_name))
+    return expired
+
+
+def make_ecs_client(region_id: str):
+    from alibabacloud_ecs20140526.client import Client as EcsClient
+    from alibabacloud_tea_openapi.models import Config as OpenApiConfig
+
+    access_key_id = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID", "")
+    access_key_secret = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET", "")
+    if not access_key_id or not access_key_secret:
+        raise SystemExit(
+            "ALIBABA_CLOUD_ACCESS_KEY_ID/SECRET are empty; in CI they come from the "
+            "ALICLOUD_ECS_ACCESS_KEY_ID/SECRET repository secrets (a missing or "
+            "misnamed secret expands to an empty string)."
+        )
+    return EcsClient(
+        OpenApiConfig(
+            access_key_id=access_key_id,
+            access_key_secret=access_key_secret,
+            region_id=region_id,
+            endpoint=f"ecs.{region_id}.aliyuncs.com",
+        )
+    )
+
+
+def delete_instance(client, instance_id: str, region_id: str | None = None) -> bool:
+    import time
+
+    from alibabacloud_ecs20140526 import models as ecs_models
+
+    # Export the serial console before destroying the evidence: a cancelled
+    # run still runs this teardown job, and the console tail (cloud-init,
+    # runner service logs, kernel OOM records) is the only witness of what
+    # happened on the machine.
+    if region_id is not None:
+        provision.dump_console_output(client, region_id, instance_id)
+
+    # DeleteInstance rejects instances that are still Initializing (or in
+    # another transitional status). Teardown runs right after a cancelled
+    # run, when the instance may be only seconds old, so retry the transient
+    # status errors for a few minutes instead of leaking the instance.
+    deadline = time.monotonic() + 5 * 60
+    while True:
+        try:
+            client.delete_instance(ecs_models.DeleteInstanceRequest(instance_id=instance_id, force=True))
+            print(f"Deleted instance {instance_id}", flush=True)
+            return True
+        except Exception as error:  # noqa: BLE001
+            message = str(error)
+            if "InvalidInstanceId.NotFound" in message:
+                print(f"Instance {instance_id} already gone", flush=True)
+                return True
+            if "IncorrectInstanceStatus" in message and time.monotonic() < deadline:
+                print(f"Instance {instance_id} is in a transitional status; retrying delete", flush=True)
+                time.sleep(15)
+                continue
+            print(f"Failed to delete instance {instance_id}: {error}", flush=True)
+            return False
+
+
+def deregister_runner(token: str, repo: str, runner_name: str) -> bool:
+    runner = provision.find_runner_by_name(token, repo, runner_name)
+    if runner is None:
+        print(f"Runner {runner_name} is not registered", flush=True)
+        return True
+    try:
+        provision.github_api(token, "DELETE", f"/repos/{repo}/actions/runners/{runner['id']}")
+        print(f"Deregistered runner {runner_name} (id {runner['id']})", flush=True)
+        return True
+    except Exception as error:  # noqa: BLE001
+        print(f"Failed to deregister runner {runner_name}: {error}", flush=True)
+        return False
+
+
+def list_managed_instances(client, region_id: str) -> list[tuple[str, str, str]]:
+    from alibabacloud_ecs20140526 import models as ecs_models
+
+    result: list[tuple[str, str, str]] = []
+    next_token = None
+    while True:
+        request = ecs_models.DescribeInstancesRequest(
+            region_id=region_id,
+            tag=[
+                ecs_models.DescribeInstancesRequestTag(
+                    key=provision.MANAGED_BY_TAG_KEY, value=provision.MANAGED_BY_TAG_VALUE
+                )
+            ],
+            max_results=100,
+            next_token=next_token,
+        )
+        response = client.describe_instances(request)
+        for instance in response.body.instances.instance:
+            result.append((instance.instance_id, instance.instance_name, instance.creation_time))
+        next_token = response.body.next_token
+        if not next_token:
+            return result
+
+
+def sweep(client, region_id: str, repo: str, github_token: str, ttl: timedelta) -> int:
+    instances = list_managed_instances(client, region_id)
+    print(f"Found {len(instances)} managed instance(s) in {region_id}", flush=True)
+    expired = expired_instance_names(instances, datetime.now(timezone.utc), ttl)
+    ok = True
+    for instance_id, instance_name in expired:
+        print(f"Instance {instance_id} ({instance_name}) exceeds TTL {ttl}; deleting", flush=True)
+        # Runner names mirror instance names by construction in the provision
+        # script (both are qreg-ecs-<run_id>).
+        ok &= delete_instance(client, instance_id, region_id)
+        ok &= deregister_runner(github_token, repo, instance_name)
+    return 0 if ok else 1
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--region-id", default=os.environ.get("ALIYUN_ECS_REGION_ID"))
+    parser.add_argument("--repo", default=os.environ.get("GITHUB_REPOSITORY"))
+    parser.add_argument("--github-token", default=os.environ.get("GH_PERSONAL_ACCESS_TOKEN"))
+    parser.add_argument("--instance-id", default=os.environ.get("QUERY_REGRESSION_ECS_INSTANCE_ID"))
+    parser.add_argument("--runner-name", default=os.environ.get("QUERY_REGRESSION_ECS_RUNNER_NAME"))
+    parser.add_argument(
+        "--sweep",
+        action="store_true",
+        help="Janitor mode: delete all managed instances older than --sweep-ttl-hours.",
+    )
+    parser.add_argument("--sweep-ttl-hours", type=float, default=SWEEP_TTL.total_seconds() / 3600)
+    args = parser.parse_args()
+
+    for name in ("region_id", "repo", "github_token"):
+        if not getattr(args, name):
+            raise SystemExit(f"Missing required configuration: --{name.replace('_', '-')}")
+
+    client = make_ecs_client(args.region_id)
+
+    if args.sweep:
+        return sweep(
+            client, args.region_id, args.repo, args.github_token, timedelta(hours=args.sweep_ttl_hours)
+        )
+
+    ok = True
+    if args.instance_id:
+        ok &= delete_instance(client, args.instance_id, args.region_id)
+    else:
+        print("No instance id given; skipping instance deletion", flush=True)
+    if args.runner_name:
+        ok &= deregister_runner(args.github_token, args.repo, args.runner_name)
+    else:
+        print("No runner name given; skipping runner deregistration", flush=True)
+    return 0 if ok else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,144 @@
+// Warns when a repository member has too many open pull requests.
+//
+// Review capacity is the bottleneck, so authors are expected to land or close
+// existing work before opening more. Drafts are not counted and never warned
+// about: they are not asking for review yet.
+// This is advisory only — it never closes a pull request or fails the job.
+
+import { appendFileSync } from "node:fs";
+
+const MARKER = "<!-- pr-open-limit -->";
+// GITHUB_TOKEN can only edit comments authored by the Actions bot itself.
+const COMMENT_AUTHOR = "github-actions[bot]";
+// Repository permissions that mark someone as part of the team.
+const TEAM_PERMISSIONS = ["admin", "maintain", "write"];
+// Keep the comment readable for authors who are far over the limit.
+const MAX_LISTED_PRS = 10;
+
+function summary(text) {
+  if (process.env.GITHUB_STEP_SUMMARY) {
+    appendFileSync(process.env.GITHUB_STEP_SUMMARY, `${text}\n`);
+  }
+  console.log(text);
+}
+
+// `author_association` cannot be used here: it is computed from what the caller
+// is allowed to see, so GITHUB_TOKEN reports a private organization member as
+// CONTRIBUTOR. Repository permission is viewer-independent.
+//
+// On error this returns true rather than false. A token that cannot read
+// permissions would otherwise make the whole check silently pass everyone.
+async function isTeamMember(octokit, owner, repo, username) {
+  try {
+    const { data } = await octokit.repos.getCollaboratorPermissionLevel({
+      owner,
+      repo,
+      username,
+    });
+    // Never log the level itself: job logs are public.
+    return TEAM_PERMISSIONS.includes(data.permission);
+  } catch (error) {
+    console.log(
+      `Cannot read repository permission for \`${username}\` (HTTP ${error.status}); applying the limit anyway.`
+    );
+    return true;
+  }
+}
+
+function buildComment(author, total, limit, openPrs) {
+  const listed = openPrs
+    .slice(0, MAX_LISTED_PRS)
+    .map((pr) => `- #${pr.number} ${pr.title}`);
+  const hidden = openPrs.length - listed.length;
+  const list = hidden > 0 ? `${listed.join("\n")}\n- ...and ${hidden} more` : listed.join("\n");
+
+  return `${MARKER}
+> [!WARNING]
+> @${author} has **${total}** open pull requests in this repository, over the limit of **${limit}**.
+
+Review is the scarcest resource here. Please land or close some of these before
+pushing this one forward:
+
+${list}
+
+This check is advisory for now and blocks nothing.`;
+}
+
+async function upsertComment(octokit, params, body) {
+  const comments = await octokit.paginate(octokit.issues.listComments, {
+    ...params,
+    per_page: 100,
+  });
+  const existing = comments.find(
+    (comment) =>
+      comment.user?.login === COMMENT_AUTHOR && comment.body?.includes(MARKER)
+  );
+
+  if (existing) {
+    await octokit.issues.updateComment({
+      owner: params.owner,
+      repo: params.repo,
+      comment_id: existing.id,
+      body,
+    });
+  } else {
+    await octokit.issues.createComment({ ...params, body });
+  }
+}
+
+(async () => {
+  const { Octokit } = await import("@octokit/rest");
+
+  const [owner, repo] = process.env.GITHUB_REPOSITORY.split("/");
+  const prNumber = Number(process.env.PR_NUMBER);
+  const author = process.env.PR_AUTHOR;
+  const rawLimit = Number(process.env.MAX_OPEN_PRS);
+  const limit = Number.isInteger(rawLimit) && rawLimit >= 0 ? rawLimit : 5;
+
+  if (author.endsWith("[bot]")) {
+    summary(`Skipping bot author \`${author}\`.`);
+    return;
+  }
+
+  if (process.env.PR_IS_DRAFT === "true") {
+    summary(`Skipping #${prNumber}: draft.`);
+    return;
+  }
+
+  const octokit = new Octokit({ auth: process.env.GITHUB_TOKEN });
+
+  if (!(await isTeamMember(octokit, owner, repo, author))) {
+    summary(`Skipping \`${author}\`: not a team member.`);
+    return;
+  }
+
+  const allOpen = await octokit.paginate(octokit.pulls.list, {
+    owner,
+    repo,
+    state: "open",
+    sort: "created",
+    direction: "asc",
+    per_page: 100,
+  });
+  // The pull request that triggered this run is already open, so exclude it and
+  // report the total separately.
+  const others = allOpen.filter(
+    (pr) => pr.user?.login === author && pr.number !== prNumber && !pr.draft
+  );
+  const total = others.length + 1;
+
+  if (others.length < limit) {
+    summary(`\`${author}\` has ${total} open pull requests (limit ${limit}). OK.`);
+    return;
+  }
+
+  summary(`\`${author}\` has ${total} open pull requests, over the limit of ${limit}.`);
+  await upsertComment(
+    octokit,
+    { owner, repo, issue_number: prNumber },
+    buildComment(author, total, limit, others)
+  );
+})().catch((error) => {
+  console.error(error);
+  process.exit(1);
+});
@@ -92,7 +92,7 @@ function thresholdStatus(thresholds, query) {
  const hits = (Array.isArray(thresholds) ? thresholds : [])
    .filter(item => query === undefined || (hasScopedQuery(item) && String(item.query) === query))
    .map(formatThreshold);
-  return hits.length > 0 ? hits.join(', ') : 'N/A';
+  return hits.length > 0 ? hits.join('; ') : 'N/A';
 }

 function hasScopedQuery(threshold) {
@@ -115,7 +115,7 @@ function formatThreshold(threshold) {
  return `${name}${scope.length > 0 ? ` [${scope.join(', ')}]` : ''}: ${status}${reason}`;
 }

-function syntheticThresholdStatus(thresholds, measurementNames) {
+function classifyThresholds(thresholds, measurementNames) {
  const unscoped = [];
  const unmatched = new Map();
  for (const threshold of Array.isArray(thresholds) ? thresholds : []) {
@@ -130,7 +130,11 @@ function syntheticThresholdStatus(thresholds, measurementNames) {
      unmatched.set(query, entries);
    }
  }
+  return { unscoped, unmatched };
+}

+function syntheticThresholdStatus(thresholds, measurementNames) {
+  const { unscoped, unmatched } = classifyThresholds(thresholds, measurementNames);
  const parts = [];
  if (unscoped.length > 0) {
    parts.push(`case/storage threshold: ${thresholdStatus(unscoped)}`);
@@ -141,6 +145,21 @@ function syntheticThresholdStatus(thresholds, measurementNames) {
  return parts.length > 0 ? parts.join('; ') : 'N/A';
 }

+// Case-level (query-unscoped) thresholds formatted for the collapsible details
+// block below the summary table: individual items, `; `-separated, without the
+// `case/storage threshold:` label (the section heading already says that).
+function syntheticThresholdDetail(thresholds, measurementNames) {
+  const { unscoped, unmatched } = classifyThresholds(thresholds, measurementNames);
+  const items = [];
+  if (unscoped.length > 0) {
+    items.push(thresholdStatus(unscoped));
+  }
+  for (const query of Array.from(unmatched.keys()).sort()) {
+    items.push(`unmatched query ${query}: ${thresholdStatus(unmatched.get(query))}`);
+  }
+  return items.length > 0 ? items.join('; ') : 'N/A';
+}
+
 function joinDetails(...details) {
  const present = details.filter(detail => detail && detail !== 'N/A');
  return present.length > 0 ? present.join('; ') : 'N/A';
@@ -229,7 +248,7 @@ function collectReportRows(report, reportPath) {
      ),
    };
  });
-  const syntheticThresholds = syntheticThresholdStatus(thresholds, measurementNames);
+  const syntheticThresholds = syntheticThresholdDetail(thresholds, measurementNames);
  if (syntheticThresholds !== 'N/A') {
    rows.push({
      caseName: name,
@@ -239,22 +258,46 @@ function collectReportRows(report, reportPath) {
      candidateMedian: 'N/A',
      regression: 'N/A',
      threshold: syntheticThresholds,
+      kind: 'case-thresholds',
    });
  }
  return rows;
 }

+function renderThresholdDetails(rows) {
+  const entries = rows.filter(row => row.kind === 'case-thresholds');
+  if (entries.length === 0) {
+    return '';
+  }
+  // The <details>/<summary> tags are emitted literally; only the entry content
+  // goes through text() (so `[ ] ( ) |` etc. stay escaped while the tags render).
+  const lines = [
+    '<details><summary>Case / storage thresholds</summary>',
+    '',
+  ];
+  for (const row of entries) {
+    lines.push(`- ${text(row.caseName)}: ${text(row.threshold)}`);
+  }
+  lines.push('</details>');
+  return lines.join('\n');
+}
+
 function renderSummaryTable(rows) {
  const lines = [
    '| Case | Query | Case status | Base median ms | Candidate median ms | Regression | Threshold |',
    '| --- | --- | --- | ---: | ---: | ---: | --- |',
  ];
  for (const row of rows) {
+    if (row.kind === 'case-thresholds') {
+      continue;
+    }
    lines.push(
-      `| ${text(row.caseName)} | ${text(row.query)} | ${statusEmoji(row.status)} \`${text(row.status)}\` | ${text(row.baseMedian)} | ${text(row.candidateMedian)} | ${text(row.regression)} | ${text(row.threshold)} |`
+      `| ${text(row.caseName)} | ${text(row.query)} | ${statusEmoji(row.status)} ${text(row.status)} | ${text(row.baseMedian)} | ${text(row.candidateMedian)} | ${text(row.regression)} | ${text(row.threshold)} |`
    );
  }
-  return lines.join('\n');
+  const details = renderThresholdDetails(rows);
+  const table = lines.join('\n');
+  return details === '' ? table : `${table}\n\n${details}`;
 }

 module.exports = async function validateQueryRegressionComment({ github, context, core }) {
@@ -187,11 +187,11 @@ test('sorts the base and candidate query union and aggregates scoped thresholds'
  assert.equal(rows[0].threshold, 'base measurement missing');
  assert.equal(
    rows[1].threshold,
-    'candidate measurement missing; p95 [target=base]: warn, absolute [target=candidate, encoding=plain]: pass'
+    'candidate measurement missing; p95 [target=base]: warn; absolute [target=candidate, encoding=plain]: pass'
  );
 });

-test('preserves unscoped and unmatched thresholds in a synthetic N/A row', () => {
+test('keeps unscoped and unmatched thresholds in a case-thresholds detail row', () => {
  const rows = collectReportRows(report('thresholds', {
    base: [{ name: 'measured', latency_ms_median: 10 }],
    candidate: [{ name: 'measured', latency_ms_median: 11 }],
@@ -214,9 +214,10 @@ test('preserves unscoped and unmatched thresholds in a synthetic N/A row', () =>
  assert.equal(rows[0].query, 'measured');
  assert.equal(rows[0].threshold, 'query limit [target=base]: passed');
  assert.equal(rows[1].query, 'N/A');
+  assert.equal(rows[1].kind, 'case-thresholds');
  assert.equal(
    rows[1].threshold,
-    'case/storage threshold: min_files [target=base]: passed, min_files [target=candidate]: failed, encoding limit [target=candidate, encoding=plain]: failed; unmatched query not-measured: orphaned limit [target=base, encoding=json]: failed (reason: measurement unavailable)'
+    'min_files [target=base]: passed; min_files [target=candidate]: failed; encoding limit [target=candidate, encoding=plain]: failed; unmatched query not-measured: orphaned limit [target=base, encoding=json]: failed (reason: measurement unavailable)'
  );
 });

@@ -239,12 +240,50 @@ test('keeps unscoped thresholds out of undefined and null query rows', () => {
  assert.equal(byQuery.get('undefined').threshold, 'N/A');
  assert.equal(byQuery.get('null').threshold, 'N/A');
  assert.equal(rows.filter(row => row.query === 'N/A').length, 1);
+  assert.equal(byQuery.get('N/A').kind, 'case-thresholds');
  assert.equal(
    byQuery.get('N/A').threshold,
-    'case/storage threshold: min_files [target=base]: passed, min_files [target=candidate]: failed'
+    'min_files [target=base]: passed; min_files [target=candidate]: failed'
  );
 });

+test('renders case/storage thresholds in a collapsible details block below the table', () => {
+  const rows = [
+    ...collectReportRows(report('first', {
+      base: [{ name: 'q1', latency_ms_median: 10 }],
+      candidate: [{ name: 'q1', latency_ms_median: 11 }],
+    }, [
+      { query: 'q1', threshold: 'query limit', target: 'base', status: 'passed' },
+      { threshold: 'min_files', target: 'base', status: 'passed' },
+      { threshold: 'min_files', target: 'candidate', status: 'failed' },
+    ]), '/reports/first/query-regression-report.json'),
+    ...collectReportRows({
+      case: { name: 'broken' },
+      status: 'failed',
+      error: 'connection refused',
+    }, '/reports/broken/query-regression-report.json'),
+  ];
+
+  const rendered = renderSummaryTable(rows);
+  const detailsIndex = rendered.indexOf('<details>');
+  const table = rendered.slice(0, detailsIndex);
+  const details = rendered.slice(detailsIndex);
+
+  // Query rows and abnormal N/A rows stay in the main table.
+  assert.match(table, /\| first \| q1 \| ✅ ok \|/);
+  assert.match(table, /\| broken \| N\/A \| ❌ failed \|/);
+  // Storage thresholds are not main-table rows.
+  assert.doesNotMatch(table, /^\| first \| N\/A \|/m);
+  assert.doesNotMatch(table, /min_files/);
+  // Status column is plain text without code-span backticks.
+  assert.doesNotMatch(table, /`/);
+  // Details block keeps its tags unescaped and its items '; '-separated.
+  assert.ok(details.startsWith('<details><summary>Case / storage thresholds</summary>'));
+  assert.ok(details.includes('- first: min_files \\[target=base\\]: passed; min_files \\[target=candidate\\]: failed'));
+  assert.ok(details.endsWith('</details>'));
+  assert.ok(!rendered.includes('&lt;details&gt;'));
+});
+
 test('escapes Markdown table content, including bare carriage returns', () => {
  const table = renderSummaryTable([{
    caseName: 'safe\r| injected |\n<!-- hidden -->@user <tag>',
@@ -0,0 +1,270 @@
+#!/usr/bin/env python3
+# Copyright 2023 Greptime Team
+#
+# 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.
+
+"""Resolve consecutive successful Nightly Build SHAs for query-regression.
+
+Query regression builds both binaries from git; this script only picks the
+head SHAs of two Nightly Build workflow runs (today vs the previous success).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from typing import Any
+
+
+NIGHTLY_WORKFLOW = "nightly-build.yml"
+NIGHTLY_EVENTS = frozenset({"schedule", "workflow_dispatch"})
+
+
+@dataclass(frozen=True)
+class WorkflowRun:
+    id: int
+    head_sha: str
+    head_branch: str
+    html_url: str
+    created_at: str
+    conclusion: str
+    event: str
+
+
+@dataclass(frozen=True)
+class RefPair:
+    base: WorkflowRun | None
+    candidate: WorkflowRun | None
+    skip: bool
+    reason: str
+
+
+def parse_run(payload: dict[str, Any]) -> WorkflowRun:
+    return WorkflowRun(
+        id=int(payload["id"]),
+        head_sha=str(payload["head_sha"]),
+        head_branch=str(payload.get("head_branch") or ""),
+        html_url=str(payload.get("html_url") or ""),
+        created_at=str(payload.get("created_at") or ""),
+        conclusion=str(payload.get("conclusion") or ""),
+        event=str(payload.get("event") or ""),
+    )
+
+
+def github_get(token: str, path: str, query: dict[str, str] | None = None) -> dict[str, Any]:
+    encoded = f"?{urllib.parse.urlencode(query)}" if query else ""
+    request = urllib.request.Request(
+        f"https://api.github.com{path}{encoded}",
+        method="GET",
+        headers={
+            "Accept": "application/vnd.github+json",
+            "Authorization": f"Bearer {token}",
+            "X-GitHub-Api-Version": "2022-11-28",
+        },
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=30) as response:
+            return json.loads(response.read().decode("utf-8"))
+    except urllib.error.HTTPError as error:
+        body = error.read().decode("utf-8", "replace")
+        raise SystemExit(
+            f"GitHub API GET {path} failed: HTTP {error.code}: {body}"
+        ) from error
+
+
+def list_successful_nightly_runs(
+    token: str,
+    repo: str,
+    *,
+    branch: str | None = None,
+    per_page: int = 30,
+) -> list[WorkflowRun]:
+    query = {
+        "status": "success",
+        "per_page": str(per_page),
+    }
+    if branch:
+        query["branch"] = branch
+    payload = github_get(
+        token,
+        f"/repos/{repo}/actions/workflows/{NIGHTLY_WORKFLOW}/runs",
+        query,
+    )
+    runs = [parse_run(item) for item in payload.get("workflow_runs") or []]
+    return [
+        run
+        for run in runs
+        if run.conclusion == "success" and run.event in NIGHTLY_EVENTS and run.head_sha
+    ]
+
+
+def fetch_run(token: str, repo: str, run_id: int) -> WorkflowRun:
+    payload = github_get(token, f"/repos/{repo}/actions/runs/{run_id}")
+    return parse_run(payload)
+
+
+def select_base_and_candidate(
+    runs: list[WorkflowRun],
+    *,
+    candidate_run_id: int | None = None,
+    candidate: WorkflowRun | None = None,
+) -> RefPair:
+    """Pick candidate (newest / requested) and the previous successful run.
+
+    `runs` must be newest-first. Same-SHA consecutive nightlies are skipped
+    (no commits that day).
+    """
+    if candidate is None:
+        if candidate_run_id is not None:
+            candidate = next((run for run in runs if run.id == candidate_run_id), None)
+            if candidate is None:
+                return RefPair(
+                    None,
+                    None,
+                    True,
+                    f"candidate nightly run {candidate_run_id} was not a successful "
+                    f"{NIGHTLY_WORKFLOW} run",
+                )
+        elif runs:
+            candidate = runs[0]
+        else:
+            return RefPair(None, None, True, "no successful Nightly Build runs")
+
+    branch = candidate.head_branch
+    older = [
+        run
+        for run in runs
+        if run.id != candidate.id
+        and (not branch or not run.head_branch or run.head_branch == branch)
+        and (not candidate.created_at or run.created_at <= candidate.created_at)
+    ]
+    if not older:
+        return RefPair(
+            None,
+            candidate,
+            True,
+            "no previous successful Nightly Build to compare against",
+        )
+    base = older[0]
+    if base.head_sha.lower() == candidate.head_sha.lower():
+        return RefPair(
+            base,
+            candidate,
+            True,
+            f"previous nightly SHA {base.head_sha} matches candidate; nothing to compare",
+        )
+    return RefPair(base, candidate, False, "")
+
+
+def _override_run(sha: str) -> WorkflowRun:
+    return WorkflowRun(
+        id=0,
+        head_sha=sha,
+        head_branch="",
+        html_url="",
+        created_at="",
+        conclusion="success",
+        event="workflow_dispatch",
+    )
+
+
+def override_pair(base_ref: str, candidate_ref: str) -> RefPair:
+    return RefPair(_override_run(base_ref), _override_run(candidate_ref), False, "explicit refs")
+
+
+def write_outputs(pair: RefPair) -> None:
+    values = {
+        "skip": "true" if pair.skip else "false",
+        "reason": pair.reason,
+        "base_sha": pair.base.head_sha if pair.base else "",
+        "candidate_sha": pair.candidate.head_sha if pair.candidate else "",
+        "base_run_url": pair.base.html_url if pair.base else "",
+        "candidate_run_url": pair.candidate.html_url if pair.candidate else "",
+        "base_run_id": str(pair.base.id) if pair.base and pair.base.id else "",
+        "candidate_run_id": str(pair.candidate.id) if pair.candidate and pair.candidate.id else "",
+    }
+    output_path = os.environ.get("GITHUB_OUTPUT")
+    if output_path:
+        with open(output_path, "a", encoding="utf-8") as handle:
+            for key, value in values.items():
+                handle.write(f"{key}={value}\n")
+    for key, value in values.items():
+        print(f"{key}={value}")
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--repo", default=os.environ.get("GITHUB_REPOSITORY"))
+    parser.add_argument(
+        "--token",
+        default=os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN"),
+    )
+    parser.add_argument("--branch", default=os.environ.get("NIGHTLY_BRANCH") or "")
+    parser.add_argument(
+        "--candidate-run-id",
+        default=os.environ.get("CANDIDATE_RUN_ID") or "",
+    )
+    parser.add_argument("--base-ref", default=os.environ.get("BASE_REF") or "")
+    parser.add_argument("--candidate-ref", default=os.environ.get("CANDIDATE_REF") or "")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    base_ref = args.base_ref.strip()
+    candidate_ref = args.candidate_ref.strip()
+    if base_ref and candidate_ref:
+        write_outputs(override_pair(base_ref, candidate_ref))
+        return 0
+    if bool(base_ref) != bool(candidate_ref):
+        print("Provide both --base-ref and --candidate-ref, or neither.", file=sys.stderr)
+        return 2
+    if not args.repo or not args.token:
+        print("--repo and --token (or GITHUB_REPOSITORY / GITHUB_TOKEN) are required.", file=sys.stderr)
+        return 2
+
+    candidate_run_id = int(args.candidate_run_id) if str(args.candidate_run_id).strip() else None
+    candidate: WorkflowRun | None = None
+    branch = args.branch.strip() or None
+    if candidate_run_id is not None:
+        candidate = fetch_run(args.token, args.repo, candidate_run_id)
+        if candidate.conclusion != "success":
+            write_outputs(
+                RefPair(
+                    None,
+                    candidate,
+                    True,
+                    f"nightly run {candidate_run_id} conclusion is {candidate.conclusion}",
+                )
+            )
+            return 0
+        branch = branch or candidate.head_branch or None
+
+    runs = list_successful_nightly_runs(args.token, args.repo, branch=branch)
+    pair = select_base_and_candidate(
+        runs,
+        candidate_run_id=candidate_run_id if candidate is None else None,
+        candidate=candidate,
+    )
+    write_outputs(pair)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -18,10 +18,17 @@
 from __future__ import annotations

 import argparse
+import json
 import os
 import re
+import signal
+import socket
 import subprocess
+import time
+import urllib.request
+from dataclasses import dataclass
 from pathlib import Path
+from typing import Any


 DEFAULT_CASES = [
@@ -100,38 +107,479 @@ def append_step_summary(summary: Path) -> None:
        out.write(summary.read_text())


+@dataclass(frozen=True)
+class RunTarget:
+    name: str
+    binary: Path
+    work_dir: Path
+    http_port: int
+    grpc_port: int
+    mysql_port: int
+    postgres_port: int
+    metasrv_rpc_port: int
+    metasrv_http_port: int
+    datanode_rpc_port: int
+    datanode_http_port: int
+    datanode_data_dir: Path
+    frontend_config: Path | None = None
+
+
+def allocate_ports(n: int) -> list[int]:
+    socks = []
+    try:
+        for _ in range(n):
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            sock.bind(("127.0.0.1", 0))
+            socks.append(sock)
+        return [sock.getsockname()[1] for sock in socks]
+    finally:
+        for sock in socks:
+            sock.close()
+
+
+def make_target(
+    name: str,
+    binary: Path,
+    root: Path,
+    ports: list[int],
+    frontend_config: Path | None = None,
+) -> RunTarget:
+    work_dir = root / name
+    return RunTarget(
+        name=name,
+        binary=binary,
+        work_dir=work_dir,
+        http_port=ports[4],
+        grpc_port=ports[5],
+        mysql_port=ports[6],
+        postgres_port=ports[7],
+        metasrv_rpc_port=ports[0],
+        metasrv_http_port=ports[1],
+        datanode_rpc_port=ports[2],
+        datanode_http_port=ports[3],
+        datanode_data_dir=work_dir / "datanode-0" / "data",
+        frontend_config=frontend_config,
+    )
+
+
+def wait_health(port: int, timeout_s: float = 60.0) -> None:
+    deadline = time.monotonic() + timeout_s
+    last: Exception | None = None
+    while time.monotonic() < deadline:
+        try:
+            with urllib.request.urlopen(f"http://127.0.0.1:{port}/health", timeout=2) as response:
+                if response.status < 500:
+                    return
+        except Exception as err:  # noqa: BLE001 - retain the last health diagnostic
+            last = err
+        time.sleep(0.5)
+    raise TimeoutError(f"health check timed out on port {port}: {last}")
+
+
+def component_log_dir(target: RunTarget, name: str) -> Path:
+    return target.work_dir / "logs" / ("datanode-0" if name == "datanode" else name)
+
+
+def component_command(target: RunTarget, name: str) -> list[str]:
+    log_dir = component_log_dir(target, name)
+    if name == "metasrv":
+        return [
+            str(target.binary), "metasrv", "start",
+            "--grpc-bind-addr", f"127.0.0.1:{target.metasrv_rpc_port}",
+            "--grpc-server-addr", f"127.0.0.1:{target.metasrv_rpc_port}",
+            "--http-addr", f"127.0.0.1:{target.metasrv_http_port}",
+            "--backend", "memory-store", "--enable-region-failover", "false",
+            "--log-dir", str(log_dir),
+        ]
+    if name == "datanode":
+        return [
+            str(target.binary), "datanode", "start",
+            "--grpc-bind-addr", f"127.0.0.1:{target.datanode_rpc_port}",
+            "--grpc-server-addr", f"127.0.0.1:{target.datanode_rpc_port}",
+            "--http-addr", f"127.0.0.1:{target.datanode_http_port}",
+            "--data-home", str(target.datanode_data_dir), "--log-dir", str(log_dir),
+            "--node-id", "0", "--metasrv-addrs", f"127.0.0.1:{target.metasrv_rpc_port}",
+        ]
+    if name == "frontend":
+        command = [str(target.binary), "frontend", "start"]
+        if target.frontend_config is not None:
+            command.extend(["--config-file", str(target.frontend_config)])
+        command.extend([
+            "--metasrv-addrs", f"127.0.0.1:{target.metasrv_rpc_port}",
+            "--http-addr", f"127.0.0.1:{target.http_port}",
+            "--grpc-bind-addr", f"127.0.0.1:{target.grpc_port}",
+            "--grpc-server-addr", f"127.0.0.1:{target.grpc_port}",
+            "--mysql-addr", f"127.0.0.1:{target.mysql_port}",
+            "--postgres-addr", f"127.0.0.1:{target.postgres_port}",
+            "--log-dir", str(log_dir),
+        ])
+        return command
+    raise ValueError(f"unknown component: {name}")
+
+
+def start_component(
+    target: RunTarget,
+    name: str,
+    procs: dict[tuple[str, str], subprocess.Popen[bytes]],
+) -> None:
+    if name != "metasrv":
+        metasrv = procs.get((target.name, "metasrv"))
+        if metasrv is None or metasrv.poll() is not None:
+            raise RuntimeError("metasrv exited; memory-store metadata is no longer valid")
+    if name == "datanode":
+        target.datanode_data_dir.mkdir(parents=True, exist_ok=True)
+    logs = component_log_dir(target, name)
+    logs.mkdir(parents=True, exist_ok=True)
+    with (logs / "stdout.log").open("ab") as out, (logs / "stderr.log").open("ab") as err:
+        procs[(target.name, name)] = subprocess.Popen(
+            component_command(target, name),
+            stdout=out,
+            stderr=err,
+            start_new_session=True,
+        )
+    wait_health({
+        "metasrv": target.metasrv_http_port,
+        "datanode": target.datanode_http_port,
+        "frontend": target.http_port,
+    }[name])
+
+
+def stop_component(
+    target: RunTarget,
+    name: str,
+    procs: dict[tuple[str, str], subprocess.Popen[bytes]],
+) -> None:
+    proc = procs.pop((target.name, name), None)
+    if proc is None or proc.poll() is not None:
+        return
+    try:
+        os.killpg(proc.pid, signal.SIGTERM)
+    except ProcessLookupError:
+        return
+    try:
+        proc.wait(timeout=20)
+    except subprocess.TimeoutExpired:
+        try:
+            os.killpg(proc.pid, signal.SIGKILL)
+        except ProcessLookupError:
+            return
+        proc.wait(timeout=20)
+
+
+def restart_component(
+    target: RunTarget,
+    name: str,
+    procs: dict[tuple[str, str], subprocess.Popen[bytes]],
+) -> None:
+    stop_component(target, name, procs)
+    start_component(target, name, procs)
+
+
+def stop_all(targets: list[RunTarget], procs: dict[tuple[str, str], subprocess.Popen[bytes]]) -> None:
+    for target in reversed(targets):
+        for name in ("frontend", "datanode", "metasrv"):
+            stop_component(target, name, procs)
+
+
+def load_plan(fixture_generator: Path, case_path: Path) -> dict[str, Any]:
+    result = subprocess.run(
+        [str(fixture_generator), "plan", "--case", str(case_path)],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        raise RuntimeError(f"query_perf_fixture plan failed: {result.stderr[:2000]}")
+    return json.loads(result.stdout)
+
+
+def run_direct_case(
+    args: argparse.Namespace,
+    case_path: Path,
+    work_dir: Path,
+    base_bin: Path,
+    candidate_bin: Path,
+    fixture_generator: Path,
+    runner: Path,
+) -> int:
+    ports = allocate_ports(16)
+    targets = [
+        make_target("base", base_bin, work_dir, ports[:8]),
+        make_target("candidate", candidate_bin, work_dir, ports[8:]),
+    ]
+    for target in targets:
+        if target.work_dir.exists() and any(target.work_dir.iterdir()):
+            raise RuntimeError(f"target work_dir exists and is non-empty: {target.work_dir}")
+        target.work_dir.mkdir(parents=True, exist_ok=True)
+
+    procs: dict[tuple[str, str], subprocess.Popen[bytes]] = {}
+    try:
+        for target in targets:
+            start_component(target, "metasrv", procs)
+            start_component(target, "datanode", procs)
+            start_component(target, "frontend", procs)
+
+        prepare = [
+            str(runner), "prepare-direct", "--case", str(case_path),
+            "--fixture-generator", str(fixture_generator),
+            "--base-http-port", str(targets[0].http_port),
+            "--candidate-http-port", str(targets[1].http_port),
+            "--fixture-dir", str(work_dir / "fixture"),
+            "--output", str(work_dir / "prepare-direct.json"),
+            "--http-timeout", str(args.http_timeout),
+        ]
+        if parse_bool(args.allow_large_fixture):
+            prepare.append("--allow-large-fixture")
+        prepare_status = subprocess.run(prepare, check=False).returncode
+        if prepare_status != 0:
+            return prepare_status
+
+        for target in targets:
+            stop_component(target, "datanode", procs)
+
+        prepared = json.loads((work_dir / "prepare-direct.json").read_text(encoding="utf-8"))
+        fixtures = prepared.get("fixtures")
+        if not isinstance(fixtures, list):
+            raise RuntimeError("prepare-direct report has no fixtures array")
+        for target in targets:
+            destination = target.work_dir / "materialize-destination.toml"
+            destination.write_text(
+                f"data_home = {json.dumps(str(target.datanode_data_dir))}\n"
+                "object_store = { type = \"File\" }\n",
+                encoding="utf-8",
+            )
+            for fixture in fixtures:
+                fixture_dir = fixture.get("fixture_dir") if isinstance(fixture, dict) else None
+                if not isinstance(fixture_dir, str):
+                    raise RuntimeError("prepare-direct fixture record has no fixture_dir")
+                materialize = [
+                    str(runner), "materialize", "--fixture-dir", fixture_dir,
+                    "--destination", str(destination),
+                ]
+                materialize_status = subprocess.run(materialize, check=False).returncode
+                if materialize_status != 0:
+                    return materialize_status
+
+        for target in targets:
+            start_component(target, "datanode", procs)
+
+        measure = [
+            str(runner), "measure", "--case", str(case_path),
+            "--fixture-generator", str(fixture_generator),
+            "--base-http-port", str(targets[0].http_port),
+            "--candidate-http-port", str(targets[1].http_port),
+            "--output", str(work_dir / "query-regression-report.json"),
+            "--http-timeout", str(args.http_timeout),
+        ]
+        status = subprocess.run(measure, check=False).returncode
+        if status != 0:
+            for target in targets:
+                restart_component(target, "frontend", procs)
+            status = subprocess.run(measure, check=False).returncode
+        return status
+    finally:
+        stop_all(targets, procs)
+
+
+def run_remote_case(
+    args: argparse.Namespace,
+    case_path: Path,
+    work_dir: Path,
+    base_bin: Path,
+    candidate_bin: Path,
+    fixture_generator: Path,
+    runner: Path,
+) -> int:
+    ports = allocate_ports(16)
+    targets = [
+        make_target(
+            "base",
+            base_bin,
+            work_dir,
+            ports[:8],
+            work_dir / "base" / "frontend-prom-store.toml",
+        ),
+        make_target(
+            "candidate",
+            candidate_bin,
+            work_dir,
+            ports[8:],
+            work_dir / "candidate" / "frontend-prom-store.toml",
+        ),
+    ]
+    for target in targets:
+        if target.work_dir.exists() and any(target.work_dir.iterdir()):
+            raise RuntimeError(f"target work_dir exists and is non-empty: {target.work_dir}")
+        target.work_dir.mkdir(parents=True, exist_ok=True)
+        if target.frontend_config is None:
+            raise RuntimeError(f"remote target has no frontend config path: {target.name}")
+        render = [
+            str(runner), "render-remote-config", "--case", str(case_path),
+            "--fixture-generator", str(fixture_generator),
+            "--output", str(target.frontend_config),
+        ]
+        render_status = subprocess.run(render, check=False).returncode
+        if render_status != 0:
+            return render_status
+
+    procs: dict[tuple[str, str], subprocess.Popen[bytes]] = {}
+    report = work_dir / "query-regression-report.json"
+    try:
+        for target in targets:
+            start_component(target, "metasrv", procs)
+            start_component(target, "datanode", procs)
+            start_component(target, "frontend", procs)
+
+        prepare = [
+            str(runner), "prepare-remote", "--case", str(case_path),
+            "--fixture-generator", str(fixture_generator),
+            "--base-http-port", str(targets[0].http_port),
+            "--candidate-http-port", str(targets[1].http_port),
+            "--output", str(work_dir / "prepare-remote.json"),
+            "--http-timeout", str(args.http_timeout),
+        ]
+        prepare_status = subprocess.run(prepare, check=False).returncode
+        if prepare_status != 0:
+            return prepare_status
+
+        measure = [
+            str(runner), "measure", "--case", str(case_path),
+            "--fixture-generator", str(fixture_generator),
+            "--base-http-port", str(targets[0].http_port),
+            "--candidate-http-port", str(targets[1].http_port),
+            "--output", str(report), "--http-timeout", str(args.http_timeout),
+        ]
+        measure_status = subprocess.run(measure, check=False).returncode
+        if measure_status != 0 and not report.exists():
+            return measure_status
+
+        for target in targets:
+            stop_component(target, "datanode", procs)
+
+        finalize = [
+            str(runner), "finalize-remote", "--case", str(case_path),
+            "--fixture-generator", str(fixture_generator),
+            "--candidate-bin", str(candidate_bin),
+            "--base-data-home", str(targets[0].datanode_data_dir),
+            "--candidate-data-home", str(targets[1].datanode_data_dir),
+            "--report", str(report),
+        ]
+        finalize_status = subprocess.run(finalize, check=False).returncode
+        if finalize_status != 0:
+            return finalize_status
+        final_report = json.loads(report.read_text(encoding="utf-8"))
+        return 1 if measure_status != 0 or final_report.get("status") == "failed" else 0
+    finally:
+        stop_all(targets, procs)
+
+
+def run_otlp_case(
+    args: argparse.Namespace,
+    case_path: Path,
+    work_dir: Path,
+    base_bin: Path,
+    candidate_bin: Path,
+    fixture_generator: Path,
+    runner: Path,
+) -> int:
+    if args.otelgen_bin is None:
+        raise ValueError("--otelgen-bin is required for otlp_trace_load")
+    ports = allocate_ports(16)
+    targets = [
+        make_target("base", base_bin, work_dir, ports[:8]),
+        make_target("candidate", candidate_bin, work_dir, ports[8:]),
+    ]
+    for target in targets:
+        if target.work_dir.exists() and any(target.work_dir.iterdir()):
+            raise RuntimeError(f"target work_dir exists and is non-empty: {target.work_dir}")
+        target.work_dir.mkdir(parents=True, exist_ok=True)
+
+    procs: dict[tuple[str, str], subprocess.Popen[bytes]] = {}
+    try:
+        for target in targets:
+            try:
+                start_component(target, "metasrv", procs)
+                start_component(target, "datanode", procs)
+                start_component(target, "frontend", procs)
+                command = [
+                    str(runner), "run-otlp-target", "--case", str(case_path),
+                    "--fixture-generator", str(fixture_generator),
+                    "--otelgen-bin", str(args.otelgen_bin),
+                    "--http-port", str(target.http_port),
+                    "--target-name", target.name,
+                    "--work-dir", str(target.work_dir),
+                    "--output", str(target.work_dir / "report.json"),
+                    "--http-timeout", str(args.http_timeout),
+                ]
+                status = subprocess.run(command, check=False).returncode
+            finally:
+                stop_all([target], procs)
+            if status != 0:
+                return status
+
+        output = work_dir / "query-regression-report.json"
+        finalize = [
+            str(runner), "finalize-otlp", "--case", str(case_path),
+            "--fixture-generator", str(fixture_generator),
+            "--base-result", str(targets[0].work_dir / "report.json"),
+            "--candidate-result", str(targets[1].work_dir / "report.json"),
+            "--output", str(output),
+        ]
+        status = subprocess.run(finalize, check=False).returncode
+        if status != 0:
+            return status
+        report = json.loads(output.read_text(encoding="utf-8"))
+        return 1 if report.get("status") == "failed" else 0
+    finally:
+        stop_all(targets, procs)
+
+
 def run_case(args: argparse.Namespace, case_path: Path, work_dir: Path) -> int:
    target_dir = profile_dir(args.cargo_profile)
    base_bin = args.base_bin or args.base_src / "target" / target_dir / "greptime"
    candidate_bin = args.candidate_bin or args.candidate_src / "target" / target_dir / "greptime"
    fixture_generator = args.fixture_generator or args.candidate_src / "target" / target_dir / "query_perf_fixture"
-    cmd = [
-        "uv",
-        "run",
-        "--no-project",
-        "python",
-        str(args.candidate_src / "tests/perf/query_regression_runner.py"),
-        "--case",
-        str(case_path),
-        "--base-bin",
-        str(base_bin),
-        "--candidate-bin",
-        str(candidate_bin),
-        "--fixture-generator",
-        str(fixture_generator),
-        "--work-dir",
-        str(work_dir),
-        "--http-timeout",
-        str(args.http_timeout),
-    ]
-    if parse_bool(args.allow_large_fixture):
-        cmd.append("--allow-large-fixture")
-    if args.otelgen_bin is not None:
-        cmd.extend(["--otelgen-bin", str(args.otelgen_bin)])
-
+    runner = args.runner or args.candidate_src / "target" / target_dir / "query_regression_runner"
+    plan = load_plan(fixture_generator, case_path)
+    scenario = plan.get("scenario")
+    kind = scenario.get("kind") if isinstance(scenario, dict) else None
    print(f"::group::Query regression case: {case_path}", flush=True)
    try:
-        return subprocess.run(cmd, check=False).returncode
+        if kind == "direct_readable_sst":
+            return run_direct_case(
+                args,
+                case_path,
+                work_dir,
+                base_bin,
+                candidate_bin,
+                fixture_generator,
+                runner,
+            )
+        if kind == "prom_remote_write_then_query":
+            return run_remote_case(
+                args,
+                case_path,
+                work_dir,
+                base_bin,
+                candidate_bin,
+                fixture_generator,
+                runner,
+            )
+        if kind == "otlp_trace_load":
+            return run_otlp_case(
+                args,
+                case_path,
+                work_dir,
+                base_bin,
+                candidate_bin,
+                fixture_generator,
+                runner,
+            )
+        raise ValueError(
+            f"unsupported scenario kind {kind!r}; supported: "
+            "'direct_readable_sst', 'prom_remote_write_then_query', 'otlp_trace_load'"
+        )
    finally:
        print("::endgroup::", flush=True)

@@ -157,6 +605,27 @@ def write_summary(args: argparse.Namespace, reports: list[Path]) -> int:
    return subprocess.run(cmd, check=False).returncode


+def write_failed_report(work_dir: Path, case_path: Path, error: Exception) -> None:
+    work_dir.mkdir(parents=True, exist_ok=True)
+    path = work_dir / "query-regression-report.json"
+    try:
+        report = json.loads(path.read_text(encoding="utf-8"))
+        if not isinstance(report, dict) or not isinstance(report.get("targets"), list) or not isinstance(report.get("thresholds"), list):
+            raise ValueError("existing report is malformed")
+    except (OSError, ValueError, json.JSONDecodeError):
+        report = {
+            "case_path": str(case_path),
+            "targets": [],
+            "thresholds": [],
+        }
+    report["status"] = "failed"
+    report["error"] = repr(error)
+    path.write_text(
+        json.dumps(report, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+
+
 def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--cases", action="append", help="'all', 'heavy', or comma/space separated case paths")
@@ -170,6 +639,7 @@ def main() -> int:
        default=configured_path(os.environ.get("FIXTURE_GENERATOR")),
    )
    parser.add_argument("--otelgen-bin", type=Path, default=configured_path(os.environ.get("OTELGEN_BIN")))
+    parser.add_argument("--runner", type=Path, default=configured_path(os.environ.get("QUERY_REGRESSION_RUNNER")))
    parser.add_argument("--cargo-profile", default=os.environ.get("CARGO_PROFILE", "nightly"))
    parser.add_argument("--work-dir", default=Path("query-regression-work"), type=Path)
    parser.add_argument("--http-timeout", default=os.environ.get("HTTP_TIMEOUT", "300"))
@@ -199,7 +669,12 @@ def main() -> int:
        case_path = resolve_case_path(args.candidate_src, case)
        work_dir = args.work_dir / case_slug(case_path)
        reports.append(work_dir / "query-regression-report.json")
-        case_status = run_case(args, case_path, work_dir)
+        try:
+            case_status = run_case(args, case_path, work_dir)
+        except Exception as err:  # noqa: BLE001 - preserve an aggregate report for the summary step
+            print(f"error: query regression case {case_path}: {err}", flush=True)
+            write_failed_report(work_dir, case_path, err)
+            case_status = 1
        if case_status != 0:
            status = case_status or 1

@@ -62,6 +62,10 @@ def parse_args() -> argparse.Namespace:
        action="store_true",
        help="Pass --preserve-state to the real compat run for artifact upload.",
    )
+    parser.add_argument(
+        "--from-versions-extra",
+        help="Comma-separated extra from-versions appended to the ci.toml window (for nightly runs).",
+    )
    return parser.parse_args()


@@ -108,6 +112,28 @@ def load_versions(
    return validated


+def load_extra_versions(raw: str | None, window: list[str]) -> list[str]:
+    """Parse --from-versions-extra, skipping versions already in the window.
+
+    Each entry must match VERSION_RE; like load_versions, no version may appear
+    twice in the effective list, so window entries and duplicates are skipped.
+    Order is preserved.
+    """
+    if not raw:
+        return []
+    extra: list[str] = []
+    seen: set[str] = set()
+    for item in raw.split(","):
+        version = item.strip()
+        if VERSION_RE.fullmatch(version) is None:
+            raise SystemExit(f"Invalid compat from-versions-extra version: {version!r}")
+        if version in window or version in seen:
+            continue
+        seen.add(version)
+        extra.append(version)
+    return extra
+
+
 def check_inputs(runner: Path, to_bins_dir: Path) -> None:
    if not runner.is_file():
        raise SystemExit(f"sqlness-runner binary not found: {runner}")
@@ -253,6 +279,7 @@ def main() -> int:

    check_inputs(runner, to_bins_dir)
    from_versions = load_versions(config_path, "from_versions")
+    extra_versions = load_extra_versions(args.from_versions_extra, from_versions)
    downgrade_to_versions = load_versions(
        config_path, "downgrade_to_versions", required=False
    )
@@ -260,8 +287,10 @@ def main() -> int:
    print("Compatibility from-version window:", flush=True)
    for version in from_versions:
        print(f"  - {version}", flush=True)
+    for version in extra_versions:
+        print(f"  - {version} (nightly extra)", flush=True)

-    for from_version in from_versions:
+    for from_version in [*from_versions, *extra_versions]:
        run_for_version(
            runner=runner,
            to_bins_dir=to_bins_dir,
@@ -0,0 +1,332 @@
+#!/usr/bin/env python3
+# Copyright 2023 Greptime Team
+#
+# 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 contextlib
+import importlib.util
+import io
+import json
+import tempfile
+import unittest
+import urllib.error
+from pathlib import Path
+from unittest import mock
+
+
+SCRIPT = Path(__file__).resolve().parents[1] / "update-compat-versions.py"
+SPEC = importlib.util.spec_from_file_location("update_compat_versions", SCRIPT)
+assert SPEC is not None and SPEC.loader is not None
+UPDATER = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(UPDATER)
+
+
+class UpdateCompatVersionsTest(unittest.TestCase):
+    def write_case(self, root: Path, name: str, from_range: str) -> None:
+        case_dir = root / "tests" / "compatibility" / "cases" / name
+        case_dir.mkdir(parents=True)
+        (case_dir / "case.toml").write_text(
+            f'name = "{name}"\nfrom_range = {from_range}\n', encoding="utf-8"
+        )
+
+    def write_config(self, root: Path, versions: str, trailer: str = "") -> Path:
+        config = root / "tests" / "compatibility" / "ci.toml"
+        config.parent.mkdir(parents=True, exist_ok=True)
+        config.write_text(
+            "# Keep these comments when the window is regenerated.\n"
+            f"from_versions = {versions}\n{trailer}",
+            encoding="utf-8",
+        )
+        return config
+
+    def test_stable_tags_filter_prereleases_and_nightlies(self) -> None:
+        self.assertEqual(
+            ["v1.0.0", "v1.0.2", "v1.1.0"],
+            UPDATER.stable_tags(
+                ["v1.1.0-nightly-20260101", "v1.0.2", "v1.0.0", "v1.1.0", "v1.0.0-rc.1"]
+            ),
+        )
+
+    def test_sliding_versions_roll_over_to_two_newest_minor_lines(self) -> None:
+        self.assertEqual(
+            ["v1.9.4", "v2.0.1"],
+            UPDATER.sliding_versions(["v1.8.9", "v1.9.0", "v1.9.4", "v2.0.0", "v2.0.1"]),
+        )
+
+    def test_sliding_versions_supports_one_minor_line(self) -> None:
+        self.assertEqual(["v1.1.3"], UPDATER.sliding_versions(["v1.1.0", "v1.1.3"]))
+
+    def test_sliding_versions_rejects_no_stable_tags(self) -> None:
+        with self.assertRaisesRegex(UPDATER.CompatVersionError, "No stable release tags"):
+            UPDATER.sliding_versions(["v1.1.0-rc.1", "v1.1.0-nightly-20260101"])
+
+    def test_effective_versions_returns_only_the_sliding_window(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            self.write_case(root, "historical", '["=v1.0.2", ">=v1.1.0"]')
+            versions = UPDATER.effective_versions(["v1.0.2", "v1.1.0", "v1.1.1", "v1.1.3"])
+        self.assertEqual(["v1.0.2", "v1.1.3"], versions)
+
+    def test_nightly_versions_includes_anchors_and_sliding(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            self.write_case(root, "historical", '["=v1.0.2", ">=v1.1.0"]')
+            self.write_case(root, "more_history", '["=v1.1.0", "=v1.1.1"]')
+            versions = UPDATER.nightly_versions(
+                ["v1.0.2", "v1.1.0", "v1.1.1", "v1.1.3"],
+                root / "tests" / "compatibility" / "cases",
+            )
+        self.assertEqual(["v1.0.2", "v1.1.0", "v1.1.1", "v1.1.3"], versions)
+
+    def test_exact_anchor_syntaxes_are_normalized(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            self.write_case(
+                root,
+                "historical",
+                '[" = v1.0.2 ", " == v1.1.0 ", " v1.1.1 ", "1.1.2", "v01.01.003-rc.1"]',
+            )
+            anchors = UPDATER.extract_pinned_anchors(root / "tests" / "compatibility" / "cases")
+        self.assertEqual(
+            {"v1.0.2", "v1.1.0", "v1.1.1", "v1.1.2", "v1.1.3"}, anchors
+        )
+
+    def test_ordered_ranges_and_wildcard_are_not_anchors(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            self.write_case(
+                root,
+                "non_exact",
+                '[">= v1.0.0", "<=v1.1.0", ">v1.0.2", "< v2.0.0", "*"]',
+            )
+            anchors = UPDATER.extract_pinned_anchors(root / "tests" / "compatibility" / "cases")
+        self.assertEqual(set(), anchors)
+
+    def test_update_is_sorted_and_idempotent(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config = self.write_config(
+                root,
+                '[\n  "v1.1.0",\n  "v1.0.2",\n]',
+                "# This trailing comment must remain.\nother_setting = true\n",
+            )
+            self.write_case(root, "historical", '["=v1.0.2"]')
+            tags = ["v1.0.2", "v1.1.0", "v1.1.1", "v1.1.3"]
+            with mock.patch.object(UPDATER, "git_tags", return_value=tags):
+                self.assertEqual(0, UPDATER.main(["--repo-root", str(root), "--update"]))
+                first_update = config.read_text(encoding="utf-8")
+                self.assertEqual(0, UPDATER.main(["--repo-root", str(root), "--update"]))
+            self.assertEqual(first_update, config.read_text(encoding="utf-8"))
+        self.assertIn('from_versions = ["v1.0.2", "v1.1.3"]', first_update)
+        self.assertIn("# Keep these comments when the window is regenerated.", first_update)
+        self.assertIn("# This trailing comment must remain.\nother_setting = true\n", first_update)
+
+    def test_check_reports_a_unified_diff_for_stale_config(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            self.write_config(root, '["v1.0.2"]')
+            self.write_case(root, "historical", '["=v1.0.2"]')
+            stderr = io.StringIO()
+            with mock.patch.object(UPDATER, "git_tags", return_value=["v1.0.2", "v1.1.3"]):
+                with contextlib.redirect_stderr(stderr):
+                    result = UPDATER.main(["--repo-root", str(root), "--check"])
+        self.assertEqual(1, result)
+        self.assertIn("--- a/tests/compatibility/ci.toml", stderr.getvalue())
+        self.assertIn("+++ b/tests/compatibility/ci.toml", stderr.getvalue())
+
+    def test_rejects_duplicate_config_versions(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            config = self.write_config(Path(directory), '["v1.0.2", "v1.0.2"]')
+            with self.assertRaisesRegex(UPDATER.CompatVersionError, "Duplicate from_versions"):
+                UPDATER.load_from_versions(config)
+
+    def test_rejects_unavailable_pinned_anchor(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            self.write_case(root, "historical", '["=v1.0.2"]')
+            with self.assertRaisesRegex(UPDATER.CompatVersionError, "unavailable"):
+                UPDATER.nightly_versions(
+                    ["v1.1.0"], root / "tests" / "compatibility" / "cases"
+                )
+
+    def test_rejects_malformed_exact_constraint(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            self.write_case(root, "historical", '["=v1.0"]')
+            with self.assertRaisesRegex(UPDATER.CompatVersionError, "Malformed version"):
+                UPDATER.extract_pinned_anchors(root / "tests" / "compatibility" / "cases")
+
+    def test_rejects_malformed_bare_exact_constraint(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            self.write_case(root, "historical", '["v1.0"]')
+            with self.assertRaisesRegex(UPDATER.CompatVersionError, "Malformed version"):
+                UPDATER.extract_pinned_anchors(root / "tests" / "compatibility" / "cases")
+
+    def test_rejects_malformed_ordered_constraint(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            self.write_case(root, "historical", '[">=v1.0"]')
+            with self.assertRaisesRegex(UPDATER.CompatVersionError, "Malformed version"):
+                UPDATER.extract_pinned_anchors(root / "tests" / "compatibility" / "cases")
+
+    def test_published_only_excludes_failed_release_tags(self) -> None:
+        self.assertEqual(
+            ["v1.0.2", "v1.1.3"],
+            UPDATER.effective_versions(
+                ["v1.0.2", "v1.1.3", "v1.1.4"], published={"v1.0.2", "v1.1.3"}
+            ),
+        )
+
+    def test_published_only_empty_intersection_raises(self) -> None:
+        with self.assertRaisesRegex(UPDATER.CompatVersionError, "published release assets"):
+            UPDATER.effective_versions(["v1.0.2"], published={"v1.1.4"})
+
+    def test_required_asset_names(self) -> None:
+        self.assertEqual(
+            {
+                "greptime-linux-amd64-v1.1.4.tar.gz",
+                "greptime-linux-amd64-v1.1.4.sha256sum",
+            },
+            UPDATER.required_asset_names("v1.1.4"),
+        )
+
+    class _FakeResponse:
+        def __init__(self, payload) -> None:
+            self._payload = payload
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc) -> None:
+            return None
+
+        def read(self) -> bytes:
+            return json.dumps(self._payload).encode("utf-8")
+
+    def _release(self, tag: str, *, draft=False, prerelease=False, assets=()) -> dict:
+        return {
+            "tag_name": tag,
+            "draft": draft,
+            "prerelease": prerelease,
+            "assets": [{"name": name} for name in assets],
+        }
+
+    def test_published_release_tags_parses_and_filters(self) -> None:
+        required = {
+            "greptime-linux-amd64-v1.1.4.tar.gz",
+            "greptime-linux-amd64-v1.1.4.sha256sum",
+        }
+        page_one = [
+            self._release(f"preview-{i}", assets=required) for i in range(96)
+        ] + [
+            self._release("v1.1.4", assets=required),
+            self._release("v1.1.3", draft=True, assets=required),
+            self._release("v1.1.2", prerelease=True, assets=UPDATER.required_asset_names("v1.1.2")),
+            self._release("v1.1.1", assets={"greptime-linux-amd64-v1.1.1.tar.gz"}),
+        ]
+        self.assertEqual(100, len(page_one))
+        page_two = [
+            self._release("v1.0.2", assets=UPDATER.required_asset_names("v1.0.2")),
+            self._release("v1.0.1-beta", assets=required),
+        ]
+        calls: list[str] = []
+
+        def fake_urlopen(request, *args, **kwargs):
+            calls.append(request.full_url)
+            payload = page_two if "page=2" in request.full_url else page_one
+            return self._FakeResponse(payload)
+
+        with mock.patch.object(UPDATER.urllib.request, "urlopen", side_effect=fake_urlopen):
+            published = UPDATER.published_release_tags("GreptimeTeam", "greptimedb", None)
+        # Draft releases are excluded; prerelease releases with the required
+        # assets are included; releases missing assets or with non-stable tags
+        # are excluded.
+        self.assertEqual({"v1.1.4", "v1.1.2", "v1.0.2"}, published)
+        self.assertEqual(2, len(calls))
+        self.assertTrue(all("per_page=100" in url for url in calls))
+
+    def test_published_release_tags_http_error_raises(self) -> None:
+        with mock.patch.object(
+            UPDATER.urllib.request,
+            "urlopen",
+            side_effect=urllib.error.HTTPError(
+                "https://api.github.com/repos/GreptimeTeam/greptimedb/releases",
+                403,
+                "Forbidden",
+                None,
+                None,
+            ),
+        ):
+            with self.assertRaisesRegex(
+                UPDATER.CompatVersionError, "Unable to fetch published releases"
+            ):
+                UPDATER.published_release_tags("GreptimeTeam", "greptimedb", None)
+
+    def test_check_anchors_rejects_unpublished_anchor(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            self.write_case(root, "historical", '["=v1.0.2", "=v1.1.0"]')
+            with mock.patch.object(UPDATER, "git_tags", return_value=["v1.0.2", "v1.1.0"]):
+                with mock.patch.object(
+                    UPDATER, "published_release_tags", return_value={"v1.0.2"}
+                ):
+                    result = UPDATER.main(["--check-anchors", "--repo-root", str(root)])
+        self.assertEqual(2, result)
+
+    def test_check_anchors_passes(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            self.write_case(root, "historical", '["=v1.0.2", "=v1.1.0"]')
+            with mock.patch.object(UPDATER, "git_tags", return_value=["v1.0.2", "v1.1.0"]):
+                with mock.patch.object(
+                    UPDATER, "published_release_tags", return_value={"v1.0.2", "v1.1.0"}
+                ):
+                    result = UPDATER.main(["--check-anchors", "--repo-root", str(root)])
+        self.assertEqual(0, result)
+
+    def test_nightly_window_prints_csv(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            self.write_case(root, "historical", '["=v1.0.2"]')
+            self.write_case(root, "more_history", '["=v1.1.0", "=v1.1.1"]')
+            stdout = io.StringIO()
+            with mock.patch.object(
+                UPDATER, "git_tags", return_value=["v1.0.2", "v1.1.0", "v1.1.1", "v1.1.3"]
+            ):
+                with contextlib.redirect_stdout(stdout):
+                    result = UPDATER.main(["--nightly-window", "--repo-root", str(root)])
+        self.assertEqual(0, result)
+        self.assertEqual("v1.0.2,v1.1.0,v1.1.1,v1.1.3\n", stdout.getvalue())
+
+    def test_update_with_published_only(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config = self.write_config(root, '["v1.0.2"]')
+            self.write_case(root, "historical", '["=v1.0.2"]')
+            with mock.patch.object(
+                UPDATER, "git_tags", return_value=["v1.0.2", "v1.1.3", "v1.1.4"]
+            ):
+                with mock.patch.object(
+                    UPDATER, "published_release_tags", return_value={"v1.0.2", "v1.1.3"}
+                ):
+                    result = UPDATER.main(
+                        ["--update", "--published-only", "--repo-root", str(root)]
+                    )
+            updated = config.read_text(encoding="utf-8")
+        self.assertEqual(0, result)
+        self.assertIn('from_versions = ["v1.0.2", "v1.1.3"]', updated)
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -0,0 +1,468 @@
+#!/usr/bin/env python3
+# Copyright 2023 Greptime Team
+#
+# 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.
+
+"""Update the compatibility CI version window from local release tags and cases.
+
+The PR window keeps only the sliding window over stable release tags whose
+GitHub releases carry the sqlness compat artifacts (greptime-linux-amd64
+tar.gz and sha256sum). Exact =vX.Y.Z `from_range` anchors in case.toml are not
+retained in the PR window: they are validated with --check-anchors and
+exercised by nightly runs via --nightly-window.
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import difflib
+import json
+import os
+import re
+import subprocess
+import sys
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Iterable
+
+
+VERSION_RE = re.compile(r"^v(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)$")
+FROM_VERSIONS_RE = re.compile(
+    r"(?ms)^(?P<indent>[ \t]*)from_versions[ \t]*=[ \t]*(?P<array>\[(?:[^\]]|\n)*?\])"
+)
+FROM_VERSIONS_ASSIGNMENT_RE = re.compile(r"(?m)^[ \t]*from_versions[ \t]*=")
+FROM_RANGE_RE = re.compile(
+    r"(?ms)^[ \t]*from_range[ \t]*=[ \t]*(?P<array>\[(?:[^\]]|\n)*?\])"
+)
+FROM_RANGE_ASSIGNMENT_RE = re.compile(r"(?m)^[ \t]*from_range[ \t]*=")
+
+
+class CompatVersionError(ValueError):
+    """A compatibility version source does not satisfy the updater policy."""
+
+
+def version_key(version: str) -> tuple[int, int, int]:
+    """Return a sortable semantic-version key for a stable version tag."""
+    match = VERSION_RE.fullmatch(version)
+    if match is None:
+        raise CompatVersionError(f"Not a stable release tag: {version!r}")
+    major, minor, patch = (int(part) for part in match.groups())
+    return major, minor, patch
+
+
+def stable_tags(tags: Iterable[str]) -> list[str]:
+    """Keep only exact stable release tags, excluding prereleases and nightlies."""
+    return sorted({tag for tag in tags if VERSION_RE.fullmatch(tag)}, key=version_key)
+
+
+def sliding_versions(tags: Iterable[str]) -> list[str]:
+    """Return the latest patch in each of the two newest stable minor lines."""
+    stable = stable_tags(tags)
+    if not stable:
+        raise CompatVersionError("No stable release tags found (expected exact v<major>.<minor>.<patch> tags)")
+
+    latest_by_minor: dict[tuple[int, int], str] = {}
+    for tag in stable:
+        major, minor, _ = version_key(tag)
+        minor_line = (major, minor)
+        if minor_line not in latest_by_minor or version_key(tag) > version_key(
+            latest_by_minor[minor_line]
+        ):
+            latest_by_minor[minor_line] = tag
+
+    newest_minor_lines = sorted(latest_by_minor, reverse=True)[:2]
+    return sorted((latest_by_minor[line] for line in newest_minor_lines), key=version_key)
+
+
+def _parse_string_array(array: str, context: str) -> list[str]:
+    without_comments = "\n".join(line.split("#", 1)[0] for line in array.splitlines())
+    try:
+        values = ast.literal_eval(without_comments)
+    except (SyntaxError, ValueError) as err:
+        raise CompatVersionError(f"Malformed string array in {context}: {err}") from err
+
+    if not isinstance(values, list) or any(not isinstance(value, str) for value in values):
+        raise CompatVersionError(f"{context} must be a TOML string array")
+    return values
+
+
+def _from_versions_match(content: str, config_path: Path) -> re.Match[str]:
+    matches = list(FROM_VERSIONS_RE.finditer(content))
+    assignments = list(FROM_VERSIONS_ASSIGNMENT_RE.finditer(content))
+    if len(assignments) != 1 or len(matches) != 1:
+        raise CompatVersionError(f"{config_path} must define exactly one from_versions array")
+    match = matches[0]
+    remainder = content[match.end() :].split("\n", 1)[0].strip()
+    if remainder and not remainder.startswith("#"):
+        raise CompatVersionError(f"Malformed from_versions assignment in {config_path}")
+    return match
+
+
+def load_from_versions(config_path: Path) -> list[str]:
+    """Validate and load the current checked-in compatibility window."""
+    if not config_path.is_file():
+        raise CompatVersionError(f"Compatibility CI config not found: {config_path}")
+
+    match = _from_versions_match(config_path.read_text(encoding="utf-8"), config_path)
+    versions = _parse_string_array(match.group("array"), str(config_path))
+    if not versions:
+        raise CompatVersionError(f"{config_path} must define a non-empty from_versions list")
+
+    duplicates = {version for version in versions if versions.count(version) > 1}
+    if duplicates:
+        raise CompatVersionError(
+            f"Duplicate from_versions in {config_path}: {', '.join(sorted(duplicates))}"
+        )
+    for version in versions:
+        version_key(version)
+    return versions
+
+
+def extract_pinned_anchors(cases_dir: Path) -> set[str]:
+    """Extract exact from_range constraints that must remain in the CI window."""
+    if not cases_dir.is_dir():
+        raise CompatVersionError(f"Compatibility cases directory not found: {cases_dir}")
+    anchors: set[str] = set()
+    for case_path in sorted(cases_dir.glob("**/case.toml")):
+        content = case_path.read_text(encoding="utf-8")
+        assignments = list(FROM_RANGE_ASSIGNMENT_RE.finditer(content))
+        matches = list(FROM_RANGE_RE.finditer(content))
+        if len(assignments) != 1 or len(matches) != 1:
+            raise CompatVersionError(f"{case_path} must define exactly one from_range array")
+
+        for constraint in _parse_string_array(matches[0].group("array"), str(case_path)):
+            anchor = exact_anchor(constraint, case_path)
+            if anchor is not None:
+                anchors.add(anchor)
+    return anchors
+
+
+def _runner_version(raw: str, case_path: Path, constraint: str) -> str:
+    """Parse and normalize a version exactly like compat_case.rs Version::parse."""
+    stripped = raw.strip()
+    if stripped.startswith("v"):
+        stripped = stripped[1:]
+    core = stripped.split("-", 1)[0].split("+", 1)[0]
+    parts = core.split(".")
+    if len(parts) != 3:
+        raise CompatVersionError(
+            f"Malformed version in from_range constraint in {case_path}: {constraint!r}"
+        )
+
+    numbers: list[int] = []
+    for part in parts:
+        if not part.isascii() or not part.isdigit():
+            raise CompatVersionError(
+                f"Malformed version in from_range constraint in {case_path}: {constraint!r}"
+            )
+        value = int(part)
+        if value > 2**64 - 1:
+            raise CompatVersionError(
+                f"Version component overflows u64 in {case_path}: {constraint!r}"
+            )
+        numbers.append(value)
+    return f"v{numbers[0]}.{numbers[1]}.{numbers[2]}"
+
+
+def exact_anchor(constraint: str, case_path: Path) -> str | None:
+    """Validate a runner constraint and return its normalized exact anchor."""
+    normalized = constraint.strip()
+    if not normalized:
+        raise CompatVersionError(f"Empty from_range constraint in {case_path}")
+    if normalized == "*":
+        return None
+
+    for operator in (">=", "<=", "==", "=", ">", "<"):
+        if normalized.startswith(operator):
+            version = _runner_version(normalized[len(operator) :], case_path, constraint)
+            return version if operator in ("==", "=") else None
+
+    return _runner_version(normalized, case_path, constraint)
+
+
+def required_asset_names(tag: str) -> set[str]:
+    """Return the release assets the sqlness compat runner downloads for a tag."""
+    return {
+        f"greptime-linux-amd64-{tag}.tar.gz",
+        f"greptime-linux-amd64-{tag}.sha256sum",
+    }
+
+
+def published_release_tags(owner: str, repo: str, token: str | None) -> set[str]:
+    """Return stable tags whose non-draft GitHub releases carry the required assets.
+
+    Only draft releases are skipped: prerelease-flagged releases with the
+    required assets are included, and VERSION_RE already excludes prerelease
+    version tags (e.g. v1.2.0-beta.1).
+    """
+    published: set[str] = set()
+    page = 1
+    while True:
+        url = (
+            f"https://api.github.com/repos/{owner}/{repo}/releases"
+            f"?per_page=100&page={page}"
+        )
+        headers = {
+            "Accept": "application/vnd.github+json",
+            "User-Agent": "update-compat-versions",
+        }
+        if token:
+            headers["Authorization"] = f"Bearer {token}"
+        request = urllib.request.Request(url, headers=headers)
+        try:
+            with urllib.request.urlopen(request) as response:
+                payload = json.loads(response.read().decode("utf-8"))
+        except (urllib.error.HTTPError, urllib.error.URLError, OSError, ValueError) as err:
+            raise CompatVersionError(
+                f"Unable to fetch published releases from {url}: {err}"
+            ) from err
+        if not isinstance(payload, list):
+            raise CompatVersionError(
+                f"Unexpected response from {url}: expected a JSON release list"
+            )
+
+        for release in payload:
+            if not isinstance(release, dict):
+                continue
+            if release.get("draft"):
+                continue
+            tag = release.get("tag_name")
+            if not isinstance(tag, str) or VERSION_RE.fullmatch(tag) is None:
+                continue
+            assets = release.get("assets")
+            asset_names: set[str] = set()
+            if isinstance(assets, list):
+                asset_names = {
+                    asset.get("name")
+                    for asset in assets
+                    if isinstance(asset, dict) and isinstance(asset.get("name"), str)
+                }
+            if required_asset_names(tag) <= asset_names:
+                published.add(tag)
+
+        if len(payload) < 100:
+            break
+        page += 1
+    return published
+
+
+def effective_versions(tags: Iterable[str], published: set[str] | None = None) -> list[str]:
+    """Return the PR window: the sliding window over stable release tags.
+
+    Exact =vX.Y.Z case anchors are intentionally excluded; they are validated
+    with --check-anchors and exercised by nightly runs (--nightly-window).
+    When ``published`` is given, only stable tags with published release
+    assets are considered.
+    """
+    stable = stable_tags(tags)
+    if published is not None:
+        stable = sorted(set(stable) & set(published), key=version_key)
+        if not stable:
+            raise CompatVersionError(
+                "No stable tags have published release assets "
+                "(greptime-linux-amd64 tar.gz and sha256sum); cannot compute the sliding window"
+            )
+    return sliding_versions(stable)
+
+
+def nightly_versions(
+    tags: Iterable[str], cases_dir: Path, published: set[str] | None = None
+) -> list[str]:
+    """Return the nightly window: exact anchors plus the sliding release window."""
+    anchors = extract_pinned_anchors(cases_dir)
+    window = effective_versions(tags, published)
+    available = set(stable_tags(tags))
+    if published is not None:
+        available &= set(published)
+    unavailable = sorted(anchors - available, key=version_key)
+    if unavailable:
+        raise CompatVersionError(
+            "Pinned compatibility anchor(s) are unavailable as stable git tags: "
+            + ", ".join(unavailable)
+        )
+    return sorted(anchors | set(window), key=version_key)
+
+
+def updated_config(content: str, config_path: Path, versions: list[str]) -> str:
+    """Return config content with only the from_versions assignment replaced."""
+    match = _from_versions_match(content, config_path)
+    # Validate the existing setting before replacing it, so malformed or duplicate
+    # checked-in configs never get silently repaired by automation.
+    load_from_versions(config_path)
+    rendered = (
+        f'{match.group("indent")}from_versions = ['
+        + ", ".join(f'"{version}"' for version in versions)
+        + "]"
+    )
+    return content[: match.start()] + rendered + content[match.end() :]
+
+
+def git_tags(repo_root: Path) -> list[str]:
+    try:
+        result = subprocess.run(
+            ["git", "tag", "--list"],
+            cwd=repo_root,
+            check=True,
+            text=True,
+            capture_output=True,
+        )
+    except (OSError, subprocess.CalledProcessError) as err:
+        raise CompatVersionError(f"Unable to read local git tags in {repo_root}: {err}") from err
+    return result.stdout.splitlines()
+
+
+def _owner_repo() -> tuple[str, str]:
+    """Return the (owner, repo) pair from GITHUB_REPOSITORY or the default repo."""
+    repository = os.environ.get("GITHUB_REPOSITORY", "GreptimeTeam/greptimedb")
+    owner, _, repo = repository.partition("/")
+    if not owner or not repo:
+        return "GreptimeTeam", "greptimedb"
+    return owner, repo
+
+
+def _github_token() -> str | None:
+    """Return the GitHub token from the environment, if any."""
+    return os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN")
+
+
+def check_anchors(repo_root: Path, cases_dir: Path) -> int:
+    """Validate every exact anchor is a stable git tag with a published release.
+
+    Returns 0 when all anchors are runnable; prints problems to stderr and
+    returns 2 when any anchor fails.
+    """
+    tags = set(git_tags(repo_root))
+    anchors = extract_pinned_anchors(cases_dir)
+    if not anchors:
+        return 0
+    published = published_release_tags(*_owner_repo(), _github_token())
+    problems: list[str] = []
+    for anchor in sorted(anchors, key=version_key):
+        if anchor not in tags:
+            problems.append(f"{anchor}: not a stable git tag")
+        elif anchor not in published:
+            problems.append(f"{anchor}: missing published release with required assets")
+    if problems:
+        print(
+            "error: exact from_range anchor(s) are not runnable in compatibility CI:",
+            file=sys.stderr,
+        )
+        for problem in problems:
+            print(f"  - {problem}", file=sys.stderr)
+        return 2
+    return 0
+
+
+def _path_from_repo(repo_root: Path, value: str | None, default: str) -> Path:
+    path = Path(value) if value is not None else Path(default)
+    return path if path.is_absolute() else repo_root / path
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    modes = parser.add_mutually_exclusive_group()
+    modes.add_argument("--check", action="store_true", help="Fail if ci.toml is stale.")
+    modes.add_argument("--update", action="store_true", help="Rewrite only ci.toml from_versions.")
+    modes.add_argument(
+        "--nightly-window",
+        action="store_true",
+        help="Print the comma-separated nightly window (anchors plus sliding) to stdout.",
+    )
+    parser.add_argument(
+        "--check-anchors",
+        action="store_true",
+        help="Validate exact case anchors are stable git tags with published release assets.",
+    )
+    parser.add_argument(
+        "--published-only",
+        action="store_true",
+        help=(
+            "With --check/--update, restrict the window to stable tags with "
+            "published release assets (requires network)."
+        ),
+    )
+    parser.add_argument(
+        "--repo-root",
+        default=str(Path(__file__).resolve().parents[2]),
+        help="Repository root used for default paths and git tags.",
+    )
+    parser.add_argument("--config", help="Compatibility CI TOML path, relative to --repo-root.")
+    parser.add_argument("--cases-dir", help="Compatibility cases path, relative to --repo-root.")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    repo_root = Path(args.repo_root).resolve()
+    config_path = _path_from_repo(repo_root, args.config, "tests/compatibility/ci.toml")
+    cases_dir = _path_from_repo(repo_root, args.cases_dir, "tests/compatibility/cases")
+
+    if not (args.check or args.update or args.nightly_window or args.check_anchors):
+        print(
+            "error: one of --check, --update, --nightly-window, or --check-anchors is required",
+            file=sys.stderr,
+        )
+        return 2
+
+    try:
+        if args.nightly_window:
+            versions = nightly_versions(git_tags(repo_root), cases_dir)
+            print(",".join(versions))
+            return 0
+
+        if args.check or args.update:
+            content = config_path.read_text(encoding="utf-8")
+            published = (
+                published_release_tags(*_owner_repo(), _github_token())
+                if args.published_only
+                else None
+            )
+            versions = effective_versions(git_tags(repo_root), published)
+            updated = updated_config(content, config_path, versions)
+        else:
+            content = updated = ""
+
+        if args.check:
+            if content == updated:
+                if args.check_anchors:
+                    return check_anchors(repo_root, cases_dir)
+                return 0
+            try:
+                display_path = config_path.relative_to(repo_root)
+            except ValueError:
+                display_path = config_path
+            sys.stderr.writelines(
+                difflib.unified_diff(
+                    content.splitlines(keepends=True),
+                    updated.splitlines(keepends=True),
+                    fromfile=f"a/{display_path}",
+                    tofile=f"b/{display_path}",
+                )
+            )
+            return 1
+
+        if content != updated:
+            config_path.write_text(updated, encoding="utf-8")
+
+        if args.check_anchors:
+            return check_anchors(repo_root, cases_dir)
+    except (CompatVersionError, OSError) as err:
+        print(f"error: {err}", file=sys.stderr)
+        return 2
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())