diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 00000000..417ad1b0 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,31 @@ +# Runtime artifacts that must never be baked into the image +harbor_test_agent_results/ + +# Local evaluator with 2 GB of cloned SWE-bench repos — never needed in K8s +evaluator/ +__pycache__/ +**/__pycache__/ +*.pyc +*.pyo +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.venv/ + +# Local dev / secrets +.env +**/.env +*.env.local + +# Dev scripts and notebooks +hack/ +docs/ + +# .git is excluded: COMMIT_HASH is injected as GIT_COMMIT build arg at image build time. +# For local dev without a built image, utils/git.py falls back to git rev-parse HEAD. +.git + +# Editor +.cursor/ +.vscode/ +*.swp diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml new file mode 100644 index 00000000..47bd3055 --- /dev/null +++ b/.github/workflows/build.yml @@ -0,0 +1,66 @@ +name: Build and Push Images + +on: + pull_request: + types: [opened, synchronize, reopened, ready_for_review] + push: + branches: [main] + tags: ["v*"] + +permissions: + contents: read + packages: write + +env: + REGISTRY: ghcr.io + +jobs: + build: + if: github.event_name != 'pull_request' || github.event.pull_request.draft == false + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + include: + - image: ridges-api + dockerfile: Dockerfile.api + - image: ridges-validator + dockerfile: Dockerfile.validator + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + + - name: Log in to GHCR + uses: docker/login-action@v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Extract Docker metadata + id: meta + uses: docker/metadata-action@v5 + with: + images: ${{ env.REGISTRY }}/ridgesai/${{ matrix.image }} + tags: | + type=semver,pattern={{version}} + type=semver,pattern={{major}}.{{minor}} + type=ref,event=branch + type=ref,event=pr + type=sha,prefix= + + - name: Build and push + uses: docker/build-push-action@v6 + with: + context: . + file: ${{ matrix.dockerfile }} + push: true + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + build-args: GIT_COMMIT=${{ github.sha }} + cache-from: type=gha + cache-to: type=gha,mode=max diff --git a/.gitignore b/.gitignore index 2d16ae6c..4012aa75 100644 --- a/.gitignore +++ b/.gitignore @@ -31,3 +31,8 @@ analyzer/ # OLD subnet_hotkeys_cache.json .python-version + +# Operational metadata +utils/registered_hotkeys.json + +.idea/ \ No newline at end of file diff --git a/Dockerfile.api b/Dockerfile.api new file mode 100644 index 00000000..6ef1c053 --- /dev/null +++ b/Dockerfile.api @@ -0,0 +1,28 @@ +# --- builder --- +FROM python:3.12-slim AS builder + +COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv + +WORKDIR /app + +# Install dependencies first (cached layer — only invalidated when lockfile changes) +COPY pyproject.toml uv.lock ./ +RUN uv sync --frozen --no-dev --no-install-project + +# Install the project itself +COPY . . +RUN uv sync --frozen --no-dev + +# --- runtime --- +FROM python:3.12-slim + +WORKDIR /app + +COPY --from=builder /app /app + +ARG GIT_COMMIT=unknown +ENV GIT_COMMIT=${GIT_COMMIT} +ENV PYTHONUNBUFFERED=1 +ENV PATH="/app/.venv/bin:$PATH" + +ENTRYPOINT ["python", "-m", "api.src.main"] diff --git a/Dockerfile.validator b/Dockerfile.validator new file mode 100644 index 00000000..57912c10 --- /dev/null +++ b/Dockerfile.validator @@ -0,0 +1,38 @@ +# --- builder --- +FROM python:3.12-slim AS builder + +RUN apt-get update && apt-get install -y --no-install-recommends git && rm -rf /var/lib/apt/lists/* + +COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv + +WORKDIR /app + +# Install dependencies first (cached layer — only invalidated when lockfile changes) +COPY pyproject.toml uv.lock ./ +RUN uv sync --frozen --no-dev --no-install-project + +# Install the project itself +COPY . . +RUN uv sync --frozen --no-dev + +# --- runtime --- +FROM python:3.12-slim + +RUN apt-get update && apt-get install -y --no-install-recommends git && rm -rf /var/lib/apt/lists/* + +RUN useradd --create-home --shell /bin/bash ridges + +WORKDIR /app + +COPY --from=builder /app /app + +RUN chown -R ridges:ridges /app + +USER ridges + +ARG GIT_COMMIT=unknown +ENV GIT_COMMIT=${GIT_COMMIT} +ENV PYTHONUNBUFFERED=1 +ENV PATH="/app/.venv/bin:$PATH" + +ENTRYPOINT ["python", "-m", "validator.main"] diff --git a/Makefile b/Makefile index fc0dc690..de239895 100644 --- a/Makefile +++ b/Makefile @@ -40,3 +40,102 @@ migrations-history: $(_DB_GUARDS) ## Show the current applied revision. migrations-current: $(_DB_GUARDS) $(_DB_ENV) uv run alembic current + +# --------------------------------------------------------------------------- +# Kubernetes local dev (kind) +# +# Prerequisites: kind, kubectl, docker, kustomize +# Usage: +# make k8s-setup — full cluster from scratch (run once) +# make k8s-images — rebuild + reload validator image after code changes +# make k8s-network — re-apply manifests after editing k8s/ files +# make k8s-screener — run screener locally against the kind cluster +# make k8s-down — destroy the kind cluster +# --------------------------------------------------------------------------- + +K8S_CLUSTER := ridges +K8S_NS := ridges +K8S_CTX := kind-$(K8S_CLUSTER) +KUBECTL := kubectl --context=$(K8S_CTX) + +.PHONY: k8s-up k8s-down k8s-images k8s-network k8s-setup k8s-screener + +## Create kind cluster with Calico CNI and ridges namespace (skip if already running) +k8s-up: + @if kind get clusters 2>/dev/null | grep -qx '$(K8S_CLUSTER)'; then \ + echo "Kind cluster '$(K8S_CLUSTER)' already exists — reusing"; \ + else \ + kind create cluster --name $(K8S_CLUSTER) --config k8s/kind-config.yaml; \ + $(KUBECTL) apply -f https://raw.githubusercontent.com/projectcalico/calico/v3.27.0/manifests/calico.yaml; \ + $(KUBECTL) wait --for=condition=ready pod -l k8s-app=calico-node -n kube-system --timeout=180s; \ + fi + $(KUBECTL) create namespace $(K8S_NS) --dry-run=client -o yaml | $(KUBECTL) apply -f - + +## Destroy the kind cluster +k8s-down: + kind delete cluster --name $(K8S_CLUSTER) + +## Build and load the validator/screener image into every kind node. +## Re-run this after any code change to pick up the new image. +k8s-images: + docker build -t ridges-validator:latest -f Dockerfile.validator \ + --build-arg GIT_COMMIT=$$(git rev-parse HEAD) . + kind load docker-image ridges-validator:latest --name $(K8S_CLUSTER) + +## Apply all k8s/ manifests (NetworkPolicies, RBAC, registry, dockerhost). +## Re-run this after editing any file under k8s/. +k8s-network: + kustomize build k8s/local | $(KUBECTL) apply -f - + $(KUBECTL) wait --for=condition=ready pod -l app=registry -n $(K8S_NS) --timeout=120s + +## Full setup: cluster → registry creds → images → manifests → secrets. +## Run once after k8s-up (or after k8s-down to start fresh). +k8s-setup: k8s-up + @# --- Registry credentials (idempotent, auto-generated on first run) --- + @if ! $(KUBECTL) get secret registry-htpasswd -n $(K8S_NS) >/dev/null 2>&1; then \ + echo "Generating registry credentials..."; \ + PASS=$$(python3 -c "import secrets; print(secrets.token_urlsafe(32))"); \ + HTPASSWD=$$(docker run --rm httpd:2-alpine htpasswd -bBn kaniko "$$PASS"); \ + $(KUBECTL) create secret generic registry-htpasswd \ + --from-literal=htpasswd="$$HTPASSWD" --namespace=$(K8S_NS); \ + $(KUBECTL) create secret docker-registry registry-creds \ + --docker-server=registry.$(K8S_NS).svc:5000 \ + --docker-username=kaniko --docker-password="$$PASS" \ + --namespace=$(K8S_NS); \ + $(KUBECTL) create secret generic registry-password \ + --from-literal=password="$$PASS" --namespace=$(K8S_NS); \ + else \ + echo "Registry credentials already exist — skipping"; \ + fi + @# --- Configure Kind nodes for plain-HTTP pulls from the in-cluster registry --- + @REGISTRY_IP=$$($(KUBECTL) get svc registry -n $(K8S_NS) -o jsonpath='{.spec.clusterIP}' 2>/dev/null || echo ""); \ + for node in $$(kind get nodes --name $(K8S_CLUSTER)); do \ + docker exec $$node mkdir -p /etc/containerd/certs.d/registry.$(K8S_NS).svc:5000; \ + printf '[host."http://registry.$(K8S_NS).svc:5000"]\n' | \ + docker exec -i $$node tee /etc/containerd/certs.d/registry.$(K8S_NS).svc:5000/hosts.toml > /dev/null; \ + docker exec $$node sh -c \ + "grep -q 'registry.$(K8S_NS).svc' /etc/hosts || echo '$$REGISTRY_IP registry.$(K8S_NS).svc' >> /etc/hosts"; \ + done + $(MAKE) k8s-images + $(MAKE) k8s-network + @# --- Screener secrets from validator/.env --- + @if [ ! -f validator/.env ]; then \ + echo "ERROR: validator/.env not found. Copy validator/.env.example and fill in values."; \ + exit 1; \ + fi + $(KUBECTL) create secret generic ridges-screener-secrets \ + --from-env-file=validator/.env --namespace=$(K8S_NS) \ + --dry-run=client -o yaml | $(KUBECTL) apply -f - + @echo "" + @echo "Kind cluster ready — context: $(K8S_CTX)" + @echo " Run: make k8s-screener" + @echo " Scale: $(KUBECTL) scale sts ridges-screener-1 --replicas=1 -n $(K8S_NS)" + +## Run screener locally against the kind cluster +k8s-screener: + RIDGES_ENVIRONMENT_TYPE=kubernetes \ + K8S_CONTEXT=$(K8S_CTX) \ + K8S_NAMESPACE=$(K8S_NS) \ + K8S_REGISTRY=registry.$(K8S_NS).svc:5000 \ + uv run python -m validator.main + diff --git a/api/Dockerfile b/api/Dockerfile deleted file mode 100644 index 6c4c51ca..00000000 --- a/api/Dockerfile +++ /dev/null @@ -1,14 +0,0 @@ -FROM python:3.12-slim - -RUN apt-get update && apt-get install -y --no-install-recommends git && rm -rf /var/lib/apt/lists/* - -COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv - -ENV PYTHONUNBUFFERED=1 - -WORKDIR /app - -COPY . . -RUN uv sync --frozen --no-dev - -CMD ["uv", "run", "-m", "api.src.main"] diff --git a/api/endpoints/scaling.py b/api/endpoints/scaling.py new file mode 100644 index 00000000..4d2c4755 --- /dev/null +++ b/api/endpoints/scaling.py @@ -0,0 +1,30 @@ +from fastapi import APIRouter + +from api.endpoints.validator import SESSION_ID_TO_VALIDATOR +from queries.agent import get_pending_work_counts + +router = APIRouter() + + +@router.get("/pending-work") +async def get_pending_work(): + """Return queue depths for KEDA autoscaling and screener capacity for observability. + + KEDA's metrics-api trigger reads ``screener_1_pending`` and + ``screener_2_pending`` via ``valueLocation``. The ``*_connected``, + ``*_busy``, and ``*_idle`` fields are not consumed by KEDA but are useful + for dashboards. + + No authentication — this endpoint is ClusterIP-only. + """ + counts = await get_pending_work_counts() + + for class_num in ("1", "2"): + prefix = f"screener-{class_num}-" + screeners = [v for v in SESSION_ID_TO_VALIDATOR.values() if v.hotkey.startswith(prefix)] + busy = sum(1 for s in screeners if s.current_evaluation_id is not None) + counts[f"screener_{class_num}_connected"] = len(screeners) + counts[f"screener_{class_num}_busy"] = busy + counts[f"screener_{class_num}_idle"] = len(screeners) - busy + + return counts diff --git a/api/src/main.py b/api/src/main.py index ad514707..9beb0d81 100644 --- a/api/src/main.py +++ b/api/src/main.py @@ -18,6 +18,7 @@ from api.endpoints.evaluation_sets import router as evaluation_sets_router from api.endpoints.evaluations import router as evaluations_router from api.endpoints.retrieval import router as retrieval_router +from api.endpoints.scaling import router as scaling_router from api.endpoints.scoring import router as scoring_router from api.endpoints.statistics import router as statistics_router from api.endpoints.validator import router as validator_router @@ -132,6 +133,7 @@ async def lifespan(app: FastAPI): app.include_router(evaluation_run_router, prefix="/evaluation-run") app.include_router(evaluations_router, prefix="/evaluation") app.include_router(statistics_router, prefix="/statistics") +app.include_router(scaling_router, prefix="/scaling") if __name__ == "__main__": diff --git a/docker-compose.yml b/docker-compose.yml index 4048ca7d..739b49ed 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -5,9 +5,9 @@ services: ports: - "5432:5432" environment: - POSTGRES_USER: - POSTGRES_PASSWORD: - POSTGRES_DB: + POSTGRES_USER: ridges + POSTGRES_PASSWORD: ridges123 + POSTGRES_DB: ridges volumes: - postgres_data:/var/lib/postgresql/data healthcheck: @@ -20,24 +20,31 @@ services: api: build: context: . - dockerfile: ./api/Dockerfile + dockerfile: ./Dockerfile.api ports: - "8000:8000" env_file: - ./api/.env + volumes: + - ./.git:/app/.git:ro depends_on: db: condition: service_healthy - s3mock: + s3: condition: service_started - s3mock: + s3: image: adobe/s3mock:latest environment: - - COM_ADOBE_TESTING_S3MOCK_STORE_INITIAL_BUCKETS=ridges-local + - COM_ADOBE_TESTING_S3MOCK_STORE_INITIAL_BUCKETS=ridges-local,ridges-registry + - COM_ADOBE_TESTING_S3MOCK_STORE_ROOT=/s3mockroot + - COM_ADOBE_TESTING_S3MOCK_STORE_RETAINFILESONEXIT=true ports: - "9090:9090" - "9191:9191" + volumes: + - s3_data:/s3mockroot volumes: postgres_data: + s3_data: diff --git a/execution/artifacts.py b/execution/artifacts.py index 28f784fc..14ad60cc 100644 --- a/execution/artifacts.py +++ b/execution/artifacts.py @@ -215,8 +215,12 @@ def parse_execution_artifacts( cost_usd = _read_proxy_cost(summary.job_dir) if summary.job_dir else None + from validator.config import RIDGES_ENVIRONMENT_TYPE + + backend_label = "harbor-k8s" if RIDGES_ENVIRONMENT_TYPE == "kubernetes" else "harbor" + return ExecutionResult( - backend="harbor", + backend=backend_label, patch=patch, verifier_reward=reward, test_results=test_results, diff --git a/execution/engine.py b/execution/engine.py index 7a28f805..2f09bef5 100644 --- a/execution/engine.py +++ b/execution/engine.py @@ -151,6 +151,7 @@ async def harbor_on_verification_started(event: Any) -> None: openrouter_config=openrouter_config, max_cost_usd=self.max_cost_usd, inference_seed=inference_seed, + fetch_task_url=fetch_task_url, on_agent_started=harbor_on_agent_started if on_agent_started is not None else None, on_verification_started=( harbor_on_verification_started if on_verification_started is not None else None diff --git a/k8s/kind-config.yaml b/k8s/kind-config.yaml new file mode 100644 index 00000000..85b73214 --- /dev/null +++ b/k8s/kind-config.yaml @@ -0,0 +1,17 @@ +--- +# kind cluster config for local testing. +# Uses Calico as the CNI so NetworkPolicies are enforced. +# +# Usage: +# kind create cluster --name ridges --config k8s/kind-config.yaml +# kubectl apply -f https://raw.githubusercontent.com/projectcalico/calico/v3.27.0/manifests/calico.yaml +kind: Cluster +apiVersion: kind.x-k8s.io/v1alpha4 +networking: + # Disable kindnet so Calico can take over + disableDefaultCNI: true + podSubnet: "192.168.0.0/16" +nodes: + - role: control-plane + - role: worker + - role: worker diff --git a/k8s/local/dockerhost.yaml b/k8s/local/dockerhost.yaml new file mode 100644 index 00000000..4346c117 --- /dev/null +++ b/k8s/local/dockerhost.yaml @@ -0,0 +1,138 @@ +--- +# TCP proxy — forwards specific ports to the Docker host machine. +# +# Pods in the ridges namespace can reach host services via: +# http://dockerhost.ridges.svc:8000 → local Ridges API +# http://dockerhost.ridges.svc:9090 → local S3mock (presigned URL host) +# +# Why hostNetwork: true? +# Calico's pod network has no route to host.docker.internal (192.168.65.254). +# Running with hostNetwork places the proxy in the Kind node's network namespace, +# which can reach 192.168.65.254 directly. No iptables DNAT is needed. +apiVersion: apps/v1 +kind: Deployment +metadata: + name: dockerhost + namespace: ridges + labels: + app: dockerhost +spec: + replicas: 1 + selector: + matchLabels: + app: dockerhost + template: + metadata: + labels: + app: dockerhost + spec: + hostNetwork: true + # Run on a fixed node so the hostNetwork pod doesn't clash with other replicas. + # For Kind there is only one 'ridges-worker' so this is a no-op but makes intent clear. + tolerations: + - operator: Exists + containers: + - name: dockerhost + image: python:3.12-slim + command: + - python3 + - -c + - | + import socket, threading, time + + # Resolve host.docker.internal once at startup (runs in host netns, works). + HOST = socket.gethostbyname('host.docker.internal') + PORTS = [8000, 9090, 9191] + print(f'Proxying ports {PORTS} -> {HOST}', flush=True) + + def forward(src, dst): + try: + while True: + d = src.recv(65536) + if not d: + break + dst.sendall(d) + except Exception: + pass + + def accept_loop(srv, port): + while True: + try: + client, _ = srv.accept() + except Exception: + return + def handle(c=client): + remote = None + try: + remote = socket.create_connection((HOST, port), timeout=10) + t = threading.Thread(target=forward, args=(c, remote), daemon=True) + t.start() + forward(remote, c) + except Exception as e: + print(f'port {port}: {e}', flush=True) + finally: + c.close() + if remote: + remote.close() + threading.Thread(target=handle, daemon=True).start() + + for port in PORTS: + srv = socket.socket() + srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + srv.bind(('0.0.0.0', port)) + srv.listen(128) + threading.Thread(target=accept_loop, args=(srv, port), daemon=True).start() + print(f'Listening on :{port}', flush=True) + + while True: + time.sleep(60) + ports: + - containerPort: 8000 + - containerPort: 9090 + - containerPort: 9191 + resources: + requests: + cpu: "50m" + memory: "32Mi" + limits: + cpu: "100m" + memory: "64Mi" +--- +apiVersion: v1 +kind: Service +metadata: + name: dockerhost + namespace: ridges + labels: + app: dockerhost +spec: + selector: + app: dockerhost + # Named ports covering the services we want to expose from the host. + # qoomon/docker-host forwards ALL ports by default, so any port below + # is reachable as dockerhost.ridges.svc:. + ports: + - name: api + port: 8000 + targetPort: 8000 + protocol: TCP + - name: s3 + port: 9090 + targetPort: 9090 + protocol: TCP + - name: s3-ssl + port: 9191 + targetPort: 9191 + protocol: TCP +--- +# DNS alias so screener pods can resolve the 's3' hostname embedded in +# presigned URLs generated by the API (S3_ENDPOINT_URL=http://s3:9090). +# CoreDNS resolves 's3' → CNAME → dockerhost.ridges.svc.cluster.local → ClusterIP. +apiVersion: v1 +kind: Service +metadata: + name: s3 + namespace: ridges +spec: + type: ExternalName + externalName: dockerhost.ridges.svc.cluster.local diff --git a/k8s/local/kustomization.yaml b/k8s/local/kustomization.yaml new file mode 100644 index 00000000..4b671f4c --- /dev/null +++ b/k8s/local/kustomization.yaml @@ -0,0 +1,5 @@ +resources: + - dockerhost.yaml + - registry.yaml + - screener-statefulsets.yaml + - network-policies.yaml diff --git a/k8s/local/network-policies.yaml b/k8s/local/network-policies.yaml new file mode 100644 index 00000000..b0aeac0d --- /dev/null +++ b/k8s/local/network-policies.yaml @@ -0,0 +1,132 @@ +--- +# Agent phase: allow only DNS + outbound port 443. +# All HTTPS traffic from the main container is transparently intercepted by the +# iptables REDIRECT rule and forwarded to the proxy SNI router on port 15443. +# The proxy sidecar (UID 1337, exempt from REDIRECT) reaches the internet directly. +# Port 80 is intentionally omitted: agents have no legitimate use for plain HTTP. +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: ridges-agent-egress + namespace: ridges +spec: + podSelector: + matchLabels: + ridges.ai/phase: agent + policyTypes: + - Egress + egress: + # DNS resolution + - ports: + - port: 53 + protocol: UDP + - port: 53 + protocol: TCP + # Outbound HTTPS only (proxy sidecar intercepts via iptables REDIRECT) + - to: + - ipBlock: + cidr: 0.0.0.0/0 + ports: + - port: 443 + protocol: TCP +--- +# Verification phase: unrestricted egress so the verifier can reach any host. +# Pods are promoted to this phase by the k8s_runtime verifier egress hook +# (label ridges.ai/phase changes from "agent" → "verification" at verification time). +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: ridges-verification-egress + namespace: ridges +spec: + podSelector: + matchLabels: + ridges.ai/phase: verification + policyTypes: + - Egress + egress: + - {} +--- +# Allow the in-cluster registry to be reached by all traffic (pods and kubelets). +# Authentication is enforced by htpasswd inside Zot — the NetworkPolicy is not +# the access-control boundary here. Restricting to a namespace selector would +# block kubelet image pulls from nodes that don't host the registry pod, which +# breaks in any multi-node topology. +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: ridges-registry-ingress + namespace: ridges +spec: + podSelector: + matchLabels: + app: registry + policyTypes: + - Ingress + ingress: + - {} +--- +# Default deny: catch-all that blocks egress for any pod not matched by a +# more-specific allow policy below. Must be listed last so that the +# explicit allows above take precedence (NetworkPolicy rules are additive, +# but the empty egress list here means "deny all" for unmatched pods). +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: default-deny-egress + namespace: ridges +spec: + podSelector: {} + policyTypes: + - Egress +--- +# Screener pods need unrestricted egress: S3, Postgres, OpenRouter, the +# in-cluster registry, and arbitrary outbound for task archiving / fetching. +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: ridges-screener-egress + namespace: ridges +spec: + podSelector: + matchLabels: + app: ridges-screener + policyTypes: + - Egress + egress: + - {} +--- +# Registry needs outbound to pull base images (DockerHub, swebench, etc.) +# during Kaniko builds. +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: ridges-registry-egress + namespace: ridges +spec: + podSelector: + matchLabels: + app: registry + policyTypes: + - Egress + egress: + - {} +--- +# Kaniko build pods need unrestricted egress: +# - DNS to resolve registry / S3 hostnames +# - S3 / S3mock to fetch build context tarballs via presigned URLs +# - In-cluster registry to push the built image +# - DockerHub / external registries to pull base images +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: ridges-build-egress + namespace: ridges +spec: + podSelector: + matchLabels: + ridges.ai/build-job: "true" + policyTypes: + - Egress + egress: + - {} diff --git a/k8s/local/registry.yaml b/k8s/local/registry.yaml new file mode 100644 index 00000000..027f3dc0 --- /dev/null +++ b/k8s/local/registry.yaml @@ -0,0 +1,157 @@ +--- +# Zot OCI registry — S3-backed, stateless, inline GC with aggressive retention. +# +# Local (Kind): S3mock on dockerhost.ridges.svc:9090, 1 replica. +# +# Deploy: +# kubectl apply -f k8s/registry.yaml +# make k8s-registry (also generates htpasswd + docker-config secrets) +# +# Credentials (auto-generated by make k8s-registry, not hardcoded here): +# registry-htpasswd — bcrypt htpasswd file for Zot auth +# registry-creds — docker-config secret for Kaniko push + kubelet pull +apiVersion: v1 +kind: ConfigMap +metadata: + name: registry-config + namespace: ridges +data: + config.json: | + { + "storage": { + "rootDirectory": "/tmp/zot", + "gc": true, + "gcDelay": "15m", + "gcInterval": "1h", + "dedupe": false, + "storageDriver": { + "name": "s3", + "rootdirectory": "/registry", + "region": "us-east-1", + "regionendpoint": "http://dockerhost.ridges.svc:9090", + "bucket": "ridges-registry", + "secure": false, + "skipverify": true, + "forcepathstyle": true + } + }, + "http": { + "address": "0.0.0.0", + "port": "5000", + "compat": ["docker2s2"], + "auth": { + "htpasswd": { + "path": "/etc/zot/htpasswd" + } + } + }, + "log": { + "level": "info" + }, + "extensions": { + "sync": { + "downloadDir": "/tmp/zot/sync", + "registries": [ + { + "urls": ["https://registry-1.docker.io"], + "onDemand": true, + "tlsVerify": true, + "maxRetries": 3, + "retryDelay": "5m", + "content": [ + { + "prefix": "swebench/**" + } + ] + } + ] + } + } + } +--- +# Zot Deployment — stateless (S3 is source of truth), 1 replica for local. +# In production, set replicas: 2+ and override ZOT_S3_ENDPOINT/credentials. +apiVersion: apps/v1 +kind: Deployment +metadata: + name: registry + namespace: ridges + labels: + app: registry +spec: + replicas: 1 # local; set to 2+ in prod + selector: + matchLabels: + app: registry + template: + metadata: + labels: + app: registry + spec: + containers: + - name: zot + image: ghcr.io/project-zot/zot-linux-amd64:v2.1.16 + args: ["serve", "/etc/zot/config.json"] + ports: + - containerPort: 5000 + name: registry + env: + # S3mock accepts any credentials; prod: use real IAM keys or workload identity. + - name: AWS_ACCESS_KEY_ID + value: "test" + - name: AWS_SECRET_ACCESS_KEY + value: "test" + volumeMounts: + - name: config + mountPath: /etc/zot/config.json + subPath: config.json + readOnly: true + - name: htpasswd + mountPath: /etc/zot/htpasswd + subPath: htpasswd + readOnly: true + # Zot temp metadata directory (required even with S3 backend). + - name: tmp + mountPath: /tmp/zot + readinessProbe: + tcpSocket: + port: 5000 + initialDelaySeconds: 5 + periodSeconds: 10 + livenessProbe: + tcpSocket: + port: 5000 + initialDelaySeconds: 15 + periodSeconds: 30 + resources: + requests: + cpu: "100m" + memory: "256Mi" + limits: + cpu: "500m" + memory: "1Gi" + volumes: + - name: config + configMap: + name: registry-config + - name: htpasswd + secret: + secretName: registry-htpasswd + - name: tmp + emptyDir: + sizeLimit: 5Gi +--- +apiVersion: v1 +kind: Service +metadata: + name: registry + namespace: ridges + labels: + app: registry +spec: + ports: + - port: 5000 + targetPort: 5000 + name: registry + selector: + app: registry diff --git a/k8s/local/screener-statefulsets.yaml b/k8s/local/screener-statefulsets.yaml new file mode 100644 index 00000000..984a3630 --- /dev/null +++ b/k8s/local/screener-statefulsets.yaml @@ -0,0 +1,281 @@ +--- +# Screener class-1 StatefulSet. +# Each Pod derives its unique SCREENER_NAME from the class + Pod ordinal: +# ridges-screener-1-0 → screener-1-0 +# ridges-screener-1-1 → screener-1-1 +# +# Scale independently (0 = idle, KEDA scales up on pending work): +# kubectl scale sts ridges-screener-1 --replicas=5 -n ridges +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: ridges-screener-1 + namespace: ridges + labels: + app: ridges-screener + ridges.ai/screener-class: "1" +spec: + replicas: 1 + serviceName: ridges-screener-1 + selector: + matchLabels: + app: ridges-screener + ridges.ai/screener-class: "1" + template: + metadata: + labels: + app: ridges-screener + ridges.ai/screener-class: "1" + spec: + # Required to create Pods (eval jobs) in the same namespace + serviceAccountName: ridges-screener + initContainers: + - name: set-screener-name + image: busybox:stable + command: + - /bin/sh + - -c + - | + ORDINAL=$(echo $HOSTNAME | rev | cut -d'-' -f1 | rev) + echo "screener-${SCREENER_CLASS}-${ORDINAL}" > /shared/screener-name + env: + - name: SCREENER_CLASS + value: "1" + volumeMounts: + - name: shared-env + mountPath: /shared + containers: + - name: screener + image: ridges-validator:latest + # Never: image is kind-loaded locally, not pulled from a registry. + imagePullPolicy: Never + command: + - /bin/sh + - -c + - | + export SCREENER_NAME=$(cat /shared/screener-name) + exec python -m validator.main + env: + - name: MODE + value: screener + - name: SCREENER_CLASS + value: "1" + - name: RIDGES_ENVIRONMENT_TYPE + value: kubernetes + - name: K8S_NAMESPACE + value: ridges + - name: K8S_REGISTRY + value: registry.ridges.svc:5000 + - name: MAX_CONCURRENT_EVALUATION_RUNS + value: "30" + - name: K8S_REGISTRY_SECRET + value: registry-creds + - name: K8S_REGISTRY_INSECURE + value: "true" # local Kind (plain HTTP); set to "false" in prod with TLS + - name: K8S_REGISTRY_PASSWORD + valueFrom: + secretKeyRef: + name: registry-password + key: password + # Override URLs from secret to use the in-cluster dockerhost proxy + # so pods can reach the host API and S3mock instead of 127.0.0.1. + - name: RIDGES_PLATFORM_URL + value: http://dockerhost.ridges.svc:8000 + - name: S3_ENDPOINT_URL + value: http://dockerhost.ridges.svc:9090 + - name: MY_POD_NAME + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: MY_POD_UID + valueFrom: + fieldRef: + fieldPath: metadata.uid + envFrom: + - secretRef: + name: ridges-screener-secrets + livenessProbe: + httpGet: + path: /healthz + port: 8080 + initialDelaySeconds: 30 + periodSeconds: 15 + readinessProbe: + httpGet: + path: /healthz + port: 8080 + initialDelaySeconds: 10 + periodSeconds: 10 + volumeMounts: + - name: shared-env + mountPath: /shared + volumes: + - name: shared-env + emptyDir: {} +--- +# Screener class-2 StatefulSet – mirrors class-1 with a different class label. +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: ridges-screener-2 + namespace: ridges + labels: + app: ridges-screener + ridges.ai/screener-class: "2" +spec: + replicas: 0 + serviceName: ridges-screener-2 + selector: + matchLabels: + app: ridges-screener + ridges.ai/screener-class: "2" + template: + metadata: + labels: + app: ridges-screener + ridges.ai/screener-class: "2" + spec: + serviceAccountName: ridges-screener + initContainers: + - name: set-screener-name + image: busybox:stable + command: + - /bin/sh + - -c + - | + ORDINAL=$(echo $HOSTNAME | rev | cut -d'-' -f1 | rev) + echo "screener-${SCREENER_CLASS}-${ORDINAL}" > /shared/screener-name + env: + - name: SCREENER_CLASS + value: "2" + volumeMounts: + - name: shared-env + mountPath: /shared + containers: + - name: screener + image: ridges-validator:latest + # Never: image is kind-loaded locally, not pulled from a registry. + imagePullPolicy: Never + command: + - /bin/sh + - -c + - | + export SCREENER_NAME=$(cat /shared/screener-name) + exec python -m validator.main + env: + - name: MODE + value: screener + - name: SCREENER_CLASS + value: "2" + - name: RIDGES_ENVIRONMENT_TYPE + value: kubernetes + - name: K8S_NAMESPACE + value: ridges + - name: K8S_REGISTRY + value: registry.ridges.svc:5000 + - name: MAX_CONCURRENT_EVALUATION_RUNS + value: "30" + - name: K8S_REGISTRY_SECRET + value: registry-creds + - name: K8S_REGISTRY_INSECURE + value: "true" # local Kind (plain HTTP); set to "false" in prod with TLS + - name: K8S_REGISTRY_PASSWORD + valueFrom: + secretKeyRef: + name: registry-password + key: password + # Override URLs from secret to use the in-cluster dockerhost proxy + # so pods can reach the host API and S3mock instead of 127.0.0.1. + - name: RIDGES_PLATFORM_URL + value: http://dockerhost.ridges.svc:8000 + - name: S3_ENDPOINT_URL + value: http://dockerhost.ridges.svc:9090 + - name: MY_POD_NAME + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: MY_POD_UID + valueFrom: + fieldRef: + fieldPath: metadata.uid + envFrom: + - secretRef: + name: ridges-screener-secrets + livenessProbe: + httpGet: + path: /healthz + port: 8080 + initialDelaySeconds: 30 + periodSeconds: 15 + readinessProbe: + httpGet: + path: /healthz + port: 8080 + initialDelaySeconds: 10 + periodSeconds: 10 + volumeMounts: + - name: shared-env + mountPath: /shared + volumes: + - name: shared-env + emptyDir: {} +--- +# Headless services for stable DNS (required by StatefulSet spec) +apiVersion: v1 +kind: Service +metadata: + name: ridges-screener-1 + namespace: ridges +spec: + clusterIP: None + selector: + app: ridges-screener + ridges.ai/screener-class: "1" +--- +apiVersion: v1 +kind: Service +metadata: + name: ridges-screener-2 + namespace: ridges +spec: + clusterIP: None + selector: + app: ridges-screener + ridges.ai/screener-class: "2" +--- +# ServiceAccount + RBAC for the screener to manage Pods and Jobs in the ridges namespace. +apiVersion: v1 +kind: ServiceAccount +metadata: + name: ridges-screener + namespace: ridges +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: ridges-screener + namespace: ridges +rules: + - apiGroups: [""] + resources: [pods, secrets] + verbs: [get, list, watch, create, delete, patch] + - apiGroups: [""] + resources: [pods/exec] + verbs: [create, get] + - apiGroups: [batch] + resources: [jobs] + verbs: [get, list, watch, create, delete] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: ridges-screener + namespace: ridges +subjects: + - kind: ServiceAccount + name: ridges-screener + namespace: ridges +roleRef: + kind: Role + name: ridges-screener + apiGroup: rbac.authorization.k8s.io diff --git a/queries/agent.py b/queries/agent.py index 7033b001..5a45ea02 100644 --- a/queries/agent.py +++ b/queries/agent.py @@ -684,3 +684,16 @@ async def get_next_agent_id_awaiting_evaluation_for_validator_hotkey( return None return result["agent_id"] + + +@db_operation +async def get_pending_work_counts(conn: DatabaseConnection) -> dict[str, int]: + row = await conn.fetchrow(""" + SELECT + (SELECT COUNT(*) FROM screener_1_queue) AS screener_1_pending, + (SELECT COUNT(*) FROM screener_2_queue) AS screener_2_pending + """) + return { + "screener_1_pending": row["screener_1_pending"], + "screener_2_pending": row["screener_2_pending"], + } diff --git a/ridges_harbor/k8s_environment.py b/ridges_harbor/k8s_environment.py new file mode 100644 index 00000000..860de7fc --- /dev/null +++ b/ridges_harbor/k8s_environment.py @@ -0,0 +1,1126 @@ +"""Kubernetes-backed Harbor environments for the Ridges screener. + +Two classes are defined here: + +* ``KubernetesEnvironment`` – generic, cloud-agnostic Harbor environment that + runs each trial inside a Kubernetes Pod. It has no Ridges-specific logic and + could eventually be contributed upstream to Harbor. + +* ``RidgesKubernetesEnvironment`` – subclass that adds: + - On-demand image building via Kaniko + an in-cluster ``registry:2`` (no + pre-building, no shared PVCs; build context is fetched from S3). + - Proxy sidecar container (MITM SSL, cost tracking, OpenRouter allow-list). + - iptables init container (``NET_ADMIN``) to transparently redirect port-443 + traffic to the proxy SNI router on port 15443 (UID 1337 is exempt so the + proxy can reach the internet directly). + - Phase labels for NetworkPolicy-based egress isolation. + - ``proxy-certs`` / ``proxy-data`` emptyDir volumes. + - ``stop()`` override that downloads ``/data`` from the Pod before deletion + so that proxy cost data is available for reporting. +""" + +from __future__ import annotations + +import asyncio +import io +import re +import shlex +import tarfile +from pathlib import Path +from typing import Any + +from harbor.environments.base import BaseEnvironment, ExecResult +from harbor.models.environment_type import EnvironmentType +from harbor.models.task.config import EnvironmentConfig +from harbor.models.trial.paths import EnvironmentPaths, TrialPaths +from kubernetes import client as k8s_client +from kubernetes import config as k8s_config +from kubernetes.client.rest import ApiException +from kubernetes.stream import stream +from tenacity import retry, stop_after_attempt, wait_exponential + +# --------------------------------------------------------------------------- +# KubernetesEnvironment – generic base +# --------------------------------------------------------------------------- + + +class KubernetesEnvironment(BaseEnvironment): + """Generic Kubernetes environment for Harbor. + + Works with any cluster accessible via a kubeconfig context or from inside + the cluster (in-cluster service-account config). No cloud-provider + dependencies. + """ + + def __init__( + self, + environment_dir: Path, + environment_name: str, + session_id: str, + trial_paths: TrialPaths, + task_env_config: EnvironmentConfig, + *, + namespace: str = "default", + image: str | None = None, + kubeconfig_context: str | None = None, + node_selector: dict[str, str] | None = None, + service_account_name: str | None = None, + memory_limit_multiplier: float | None = None, + labels: dict[str, str] | None = None, + image_pull_secrets: list[str] | None = None, + owner_pod_name: str | None = None, + owner_pod_uid: str | None = None, + **kwargs, + ): + super().__init__( + environment_dir=environment_dir, + environment_name=environment_name, + session_id=session_id, + trial_paths=trial_paths, + task_env_config=task_env_config, + **kwargs, + ) + + self.namespace = namespace + self.image = image or f"{environment_name}:latest" + self.kubeconfig_context = kubeconfig_context + self.node_selector = node_selector + self.service_account_name = service_account_name + self._extra_labels: dict[str, str] = labels or {} + self._image_pull_secrets: list[str] = image_pull_secrets or [] + self._owner_pod_name = owner_pod_name + self._owner_pod_uid = owner_pod_uid + + # Resource sizing + self.cpu_request = str(task_env_config.cpus) + self.memory_request = f"{task_env_config.memory_mb}Mi" + self.ephemeral_storage_request = f"{task_env_config.storage_mb}Mi" + + if memory_limit_multiplier is not None and memory_limit_multiplier > 0: + self.memory_limit: str | None = f"{int(task_env_config.memory_mb * memory_limit_multiplier)}Mi" + else: + self.memory_limit = None + + # Pod name: lowercase, no underscores, max 63 chars + self.pod_name = session_id.lower().replace("_", "-")[:63] + + # Kubernetes client – lazily initialised + self._core_api: k8s_client.CoreV1Api | None = None + self._batch_api: k8s_client.BatchV1Api | None = None + + # ------------------------------------------------------------------ + # BaseEnvironment abstract properties + # ------------------------------------------------------------------ + + @staticmethod + def type() -> EnvironmentType: + # There is no canonical EnvironmentType for generic k8s; return GKE so + # Harbor's existing type-enum machinery doesn't break. We always reach + # this class via import_path, not by type, so the value is never used + # for routing. + return EnvironmentType.GKE + + @property + def is_mounted(self) -> bool: + return False + + @property + def supports_gpus(self) -> bool: + return False + + @property + def can_disable_internet(self) -> bool: + # Internet isolation is handled externally via NetworkPolicies. + return True + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _validate_definition(self) -> None: + """No local file validation needed – the image lives in the registry.""" + + def _init_k8s_client(self) -> None: + """Load kubeconfig (or in-cluster config) and create API clients.""" + try: + k8s_config.load_incluster_config() + except k8s_config.ConfigException: + k8s_config.load_kube_config(context=self.kubeconfig_context) + self._core_api = k8s_client.CoreV1Api() + self._batch_api = k8s_client.BatchV1Api() + + async def _ensure_client(self) -> None: + if self._core_api is None: + await asyncio.to_thread(self._init_k8s_client) + + @property + def _api(self) -> k8s_client.CoreV1Api: + if self._core_api is None: + raise RuntimeError("Kubernetes client not initialised. Call _ensure_client() first.") + return self._core_api + + @property + def _batch(self) -> k8s_client.BatchV1Api: + if self._batch_api is None: + raise RuntimeError("Kubernetes batch client not initialised. Call _ensure_client() first.") + return self._batch_api + + # ------------------------------------------------------------------ + # Extension points (overrideable by subclasses) + # ------------------------------------------------------------------ + + def _build_labels(self) -> dict[str, str]: + base = { + "app": "ridges-eval", + "session": self.session_id[:63], + } + base.update(self._extra_labels) + return base + + def _build_volumes(self) -> list[k8s_client.V1Volume]: + return [] + + def _build_containers(self) -> list[k8s_client.V1Container]: + requests: dict[str, str] = { + "cpu": self.cpu_request, + "memory": self.memory_request, + } + if self.ephemeral_storage_request: + requests["ephemeral-storage"] = self.ephemeral_storage_request + + limits: dict[str, str] = {} + if self.memory_limit: + limits["memory"] = self.memory_limit + + return [ + k8s_client.V1Container( + name="main", + image=self.image, + command=["sleep", "infinity"], + resources=k8s_client.V1ResourceRequirements( + requests=requests, + limits=limits or None, + ), + volume_mounts=[], + ) + ] + + def _build_pod_spec(self) -> k8s_client.V1PodSpec: + spec = k8s_client.V1PodSpec( + containers=self._build_containers(), + volumes=self._build_volumes() or None, + restart_policy="Never", + ) + if self.node_selector: + spec.node_selector = self.node_selector + if self.service_account_name: + spec.service_account_name = self.service_account_name + if self._image_pull_secrets: + spec.image_pull_secrets = [k8s_client.V1LocalObjectReference(name=s) for s in self._image_pull_secrets] + return spec + + def _build_pod(self) -> k8s_client.V1Pod: + metadata = k8s_client.V1ObjectMeta( + name=self.pod_name, + namespace=self.namespace, + labels=self._build_labels(), + ) + if self._owner_pod_name and self._owner_pod_uid: + metadata.owner_references = [ + k8s_client.V1OwnerReference( + api_version="v1", + kind="Pod", + name=self._owner_pod_name, + uid=self._owner_pod_uid, + block_owner_deletion=True, + controller=False, + ), + ] + return k8s_client.V1Pod( + api_version="v1", + kind="Pod", + metadata=metadata, + spec=self._build_pod_spec(), + ) + + # ------------------------------------------------------------------ + # BaseEnvironment lifecycle + # ------------------------------------------------------------------ + + async def start(self, force_build: bool) -> None: + """Create the Pod and wait until all containers are ready.""" + await self._ensure_client() + + pod = self._build_pod() + + try: + await asyncio.to_thread( + self._api.create_namespaced_pod, + namespace=self.namespace, + body=pod, + ) + except ApiException as exc: + if exc.status == 409: + self.logger.debug(f"Pod {self.pod_name} already exists – deleting and recreating") + await self._delete_pod_and_wait() + await asyncio.to_thread( + self._api.create_namespaced_pod, + namespace=self.namespace, + body=pod, + ) + else: + raise RuntimeError(f"Failed to create Pod {self.pod_name}: {exc}") from exc + + await self._wait_for_pod_ready() + + mkdir_result = await self.exec( + f"mkdir -p {EnvironmentPaths.agent_dir} {EnvironmentPaths.verifier_dir} && " + f"chmod 777 {EnvironmentPaths.agent_dir} {EnvironmentPaths.verifier_dir}" + ) + if mkdir_result.return_code != 0: + raise RuntimeError( + f"Failed to create log directories in Pod {self.pod_name}: " + f"stdout={mkdir_result.stdout}, stderr={mkdir_result.stderr}" + ) + + async def stop(self, delete: bool = True) -> None: + """Delete the Pod (optionally).""" + if self._core_api is None: + return + if delete: + try: + await self._delete_pod_and_wait() + except RuntimeError: + self.logger.warning(f"Pod {self.pod_name} did not terminate cleanly during stop()") + + async def exec( + self, + command: str, + cwd: str | None = None, + env: dict[str, str] | None = None, + timeout_sec: int | None = None, + user: str | int | None = None, + ) -> ExecResult: + """Execute *command* inside the main container of the Pod.""" + user = self._resolve_user(user) + env = self._merge_env(env) + + await self._ensure_client() + + full_command = f"bash -c {shlex.quote(command)}" + + if env: + prefix = " ".join(f"{k}={shlex.quote(v)}" for k, v in env.items()) + full_command = f"{prefix} {full_command}" + + if cwd: + full_command = f"cd {shlex.quote(cwd)} && {full_command}" + + if user is not None: + if isinstance(user, int): + user_arg = f"$(getent passwd {user} | cut -d: -f1)" + else: + user_arg = shlex.quote(str(user)) + full_command = f"su {user_arg} -s /bin/bash -c {shlex.quote(full_command)}" + + exec_command = ["sh", "-c", full_command] + + resp = None + try: + resp = await asyncio.to_thread( + stream, + self._api.connect_get_namespaced_pod_exec, + self.pod_name, + self.namespace, + container="main", + command=exec_command, + stderr=True, + stdin=False, + stdout=True, + tty=False, + _preload_content=False, + ) + + if timeout_sec is not None: + stdout, stderr = await asyncio.wait_for( + asyncio.to_thread(self._read_exec_output, resp), + timeout=timeout_sec, + ) + else: + stdout, stderr = await asyncio.to_thread(self._read_exec_output, resp) + + resp.run_forever(timeout=0) + return_code = resp.returncode if resp.returncode is not None else 0 + return ExecResult(stdout=stdout, stderr=stderr, return_code=return_code) + + except asyncio.TimeoutError: + return ExecResult( + stdout=None, + stderr=f"Command timed out after {timeout_sec} seconds", + return_code=124, + ) + except ApiException as exc: + if exc.status == 404: + return ExecResult(stdout=None, stderr=f"Pod {self.pod_name} not found (404)", return_code=1) + return ExecResult( + stdout=None, + stderr=f"API error ({exc.status}) on Pod {self.pod_name}: {exc.reason}", + return_code=1, + ) + except Exception as exc: + return ExecResult(stdout=None, stderr=str(exc), return_code=1) + finally: + if resp is not None: + try: + resp.close() + except Exception: + pass + + @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=2, min=2, max=30), reraise=True) + async def upload_file(self, source_path: Path | str, target_path: str) -> None: + await self._ensure_client() + await self._wait_for_container_exec_ready() + + source_path = Path(source_path) + tar_buffer = io.BytesIO() + with tarfile.open(fileobj=tar_buffer, mode="w") as tar: + tar.add(str(source_path), arcname=Path(target_path).name) + tar_buffer.seek(0) + + target_dir = str(Path(target_path).parent) + await self.exec(f"mkdir -p {shlex.quote(target_dir)}", user="root") + + resp = await asyncio.to_thread( + stream, + self._api.connect_get_namespaced_pod_exec, + self.pod_name, + self.namespace, + container="main", + command=["tar", "xf", "-", "-C", target_dir], + stderr=True, + stdin=True, + stdout=True, + tty=False, + _preload_content=False, + ) + resp.write_stdin(tar_buffer.read()) + resp.run_forever(timeout=1) + resp.close() + + @retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=2, max=30), reraise=True) + async def upload_dir(self, source_dir: Path | str, target_dir: str) -> None: + await self._ensure_client() + await self._wait_for_container_exec_ready() + + source_dir = Path(source_dir) + tar_buffer = io.BytesIO() + with tarfile.open(fileobj=tar_buffer, mode="w") as tar: + for item in source_dir.rglob("*"): + if item.is_file(): + tar.add(str(item), arcname=str(item.relative_to(source_dir))) + tar_buffer.seek(0) + + await self.exec(f"mkdir -p {shlex.quote(target_dir)}", user="root") + + resp = await asyncio.to_thread( + stream, + self._api.connect_get_namespaced_pod_exec, + self.pod_name, + self.namespace, + container="main", + command=["tar", "xf", "-", "-C", target_dir], + stderr=True, + stdin=True, + stdout=True, + tty=False, + _preload_content=False, + ) + try: + resp.write_stdin(tar_buffer.read()) + except Exception as exc: + raise RuntimeError(f"Failed to write tar data to Pod {self.pod_name}: {exc}") from exc + resp.run_forever(timeout=1) + resp.close() + + @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), reraise=True) + async def download_file(self, source_path: str, target_path: Path | str) -> None: + await self._ensure_client() + target_path = Path(target_path) + target_path.parent.mkdir(parents=True, exist_ok=True) + + resp = await asyncio.to_thread( + stream, + self._api.connect_get_namespaced_pod_exec, + self.pod_name, + self.namespace, + container="main", + command=["tar", "cf", "-", source_path], + stderr=True, + stdin=False, + stdout=True, + tty=False, + _preload_content=False, + ) + + tar_data = b"" + while resp.is_open(): + resp.update(timeout=1) + if resp.peek_stdout(): + chunk = resp.read_stdout() + tar_data += chunk.encode("utf-8", errors="surrogateescape") if isinstance(chunk, str) else chunk + + tar_buffer = io.BytesIO(tar_data) + with tarfile.open(fileobj=tar_buffer, mode="r") as tar: + for member in tar.getmembers(): + if member.name == source_path or member.name.startswith(source_path.lstrip("/")): + member.name = target_path.name + tar.extract(member, path=str(target_path.parent)) + break + + @retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=2, max=30), reraise=True) + async def download_dir(self, source_dir: str, target_dir: Path | str) -> None: + await self._ensure_client() + target_dir = Path(target_dir) + target_dir.mkdir(parents=True, exist_ok=True) + + try: + resp = await asyncio.to_thread( + stream, + self._api.connect_get_namespaced_pod_exec, + self.pod_name, + self.namespace, + container="main", + command=["sh", "-c", f"cd {shlex.quote(source_dir)} && tar cf - ."], + stderr=True, + stdin=False, + stdout=True, + tty=False, + _preload_content=False, + ) + except ApiException as exc: + raise RuntimeError(f"Failed to start tar download from Pod {self.pod_name}: {exc}") from exc + + tar_data = b"" + stderr_data = "" + while resp.is_open(): + resp.update(timeout=1) + if resp.peek_stdout(): + chunk = resp.read_stdout() + tar_data += chunk.encode("utf-8", errors="surrogateescape") if isinstance(chunk, str) else chunk + if resp.peek_stderr(): + stderr_data += resp.read_stderr() + + if stderr_data and ("No such file or directory" in stderr_data or "cannot cd" in stderr_data): + raise RuntimeError(f"Failed to access directory {source_dir} in Pod {self.pod_name}: {stderr_data.strip()}") + + if not tar_data: + raise RuntimeError(f"No data received when downloading {source_dir} from Pod {self.pod_name}") + + tar_buffer = io.BytesIO(tar_data) + try: + with tarfile.open(fileobj=tar_buffer, mode="r") as tar: + tar.extractall(path=str(target_dir)) + except tarfile.TarError as exc: + raise RuntimeError(f"Failed to extract {source_dir} from Pod {self.pod_name}: {exc}") from exc + + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ + + def _read_exec_output(self, resp: Any) -> tuple[str, str]: + stdout = "" + stderr = "" + while resp.is_open(): + resp.update(timeout=1) + if resp.peek_stdout(): + stdout += resp.read_stdout() + if resp.peek_stderr(): + stderr += resp.read_stderr() + return stdout, stderr + + async def _wait_for_container_exec_ready(self, max_attempts: int = 60) -> None: + for attempt in range(max_attempts): + try: + resp = await asyncio.to_thread( + stream, + self._api.connect_get_namespaced_pod_exec, + self.pod_name, + self.namespace, + container="main", + command=["true"], + stderr=False, + stdin=False, + stdout=True, + tty=False, + _preload_content=False, + ) + resp.close() + return + except ApiException as exc: + if "container not found" in str(exc) or exc.status == 500: + if attempt % 10 == 0: + self.logger.debug(f"Container not ready, attempt {attempt + 1}/{max_attempts}") + await asyncio.sleep(3) + else: + raise + except Exception: + if attempt < max_attempts - 1: + await asyncio.sleep(3) + else: + raise + raise RuntimeError(f"Container not ready for exec after {max_attempts} attempts") + + async def _wait_for_pod_ready(self, timeout_sec: int = 300) -> None: + self.logger.debug(f"Waiting for Pod {self.pod_name} to be ready...") + for attempt in range(timeout_sec): + try: + pod = await asyncio.to_thread( + self._api.read_namespaced_pod, + name=self.pod_name, + namespace=self.namespace, + ) + phase = pod.status.phase + if phase == "Running" and pod.status.container_statuses: + if all(c.ready for c in pod.status.container_statuses): + self.logger.debug(f"Pod {self.pod_name} is ready") + return + elif phase in ("Failed", "Unknown", "Error"): + raise RuntimeError(f"Pod {self.pod_name} failed to start: {self._pod_failure_summary(pod)}") + elif phase == "Pending" and pod.status.container_statuses: + for cs in pod.status.container_statuses: + if cs.state.waiting and cs.state.waiting.reason in ("ImagePullBackOff", "ErrImagePull"): + raise RuntimeError( + f"Failed to pull image for Pod {self.pod_name}: {cs.state.waiting.message}" + ) + except ApiException as exc: + if exc.status != 404: + raise RuntimeError(f"Kubernetes API error while waiting for Pod: {exc}") from exc + + if attempt % 10 == 0: + self.logger.debug(f"Pod {self.pod_name} not ready yet ({attempt}s elapsed)") + await asyncio.sleep(1) + + raise RuntimeError(f"Pod {self.pod_name} not ready after {timeout_sec}s") + + async def _delete_pod_and_wait(self, timeout_sec: int = 60) -> None: + try: + await asyncio.to_thread( + self._api.delete_namespaced_pod, + name=self.pod_name, + namespace=self.namespace, + body=k8s_client.V1DeleteOptions(grace_period_seconds=0, propagation_policy="Foreground"), + ) + except ApiException as exc: + if exc.status == 404: + return + raise + + for _ in range(timeout_sec): + try: + await asyncio.to_thread(self._api.read_namespaced_pod, name=self.pod_name, namespace=self.namespace) + await asyncio.sleep(1) + except ApiException as exc: + if exc.status == 404: + return + + self.logger.warning(f"Pod {self.pod_name} did not terminate within {timeout_sec}s") + + def _pod_failure_summary(self, pod: Any) -> str: + parts: list[str] = [] + if pod.status.reason: + parts.append(f"reason={pod.status.reason}") + if pod.status.message: + parts.append(f"message={pod.status.message}") + if pod.status.container_statuses: + for cs in pod.status.container_statuses: + if cs.state.waiting: + parts.append(f"container {cs.name} waiting: {cs.state.waiting.reason}") + elif cs.state.terminated: + parts.append(f"container {cs.name} terminated: exit={cs.state.terminated.exit_code}") + return "; ".join(parts) or "unknown" + + +# --------------------------------------------------------------------------- +# RidgesKubernetesEnvironment – adds proxy sidecar + on-demand Kaniko builds +# --------------------------------------------------------------------------- + + +class RidgesKubernetesEnvironment(KubernetesEnvironment): + """Kubernetes environment with the Ridges proxy sidecar and on-demand image building. + + Image building flow (on cache miss): + 1. Screener creates a Kubernetes Secret with the S3 presigned URL. + 2. A Kaniko Job is created; its init container downloads the task archive + from S3 and extracts it to an emptyDir. + 3. Kaniko builds the image and pushes it to the in-cluster ``registry:2``. + 4. On success, the Pod is created to run the evaluation. + + Proxy data flow: + - The proxy sidecar writes cost/usage data to ``/data`` (emptyDir volume). + - ``stop()`` downloads ``/data`` from the Pod to ``proxy_data_dir`` on the + host before deleting the Pod so cost reporting still works. + """ + + def __init__( + self, + environment_dir: Path, + environment_name: str, + session_id: str, + trial_paths: TrialPaths, + task_env_config: EnvironmentConfig, + *, + registry: str, + task_name: str, + digest_tag: str, + task_archive_presigned_url: str, + proxy_image: str, + evaluation_run_id: str, + max_cost_usd: str = "999999", + openrouter_sidecar_env: dict[str, str] | None = None, + proxy_data_dir: str | Path | None = None, + registry_credentials_secret: str | None = None, + registry_password: str | None = None, + registry_insecure: bool = True, + **kwargs, + ): + self.registry = registry + self.task_name = task_name + self.digest_tag = digest_tag + self.task_archive_presigned_url = task_archive_presigned_url + self.proxy_image = proxy_image + self.evaluation_run_id = evaluation_run_id + self.max_cost_usd = max_cost_usd + self.openrouter_sidecar_env: dict[str, str] = openrouter_sidecar_env or {} + self.proxy_data_dir: Path | None = Path(proxy_data_dir) if proxy_data_dir else None + self.registry_credentials_secret = registry_credentials_secret + self._registry_password = registry_password + self._registry_insecure = registry_insecure + + image = f"{registry}/{task_name}:{digest_tag}" + # Eval pods need credentials to pull the task image from the in-cluster registry. + pull_secrets = [registry_credentials_secret] if registry_credentials_secret else [] + super().__init__( + environment_dir=environment_dir, + environment_name=environment_name, + session_id=session_id, + trial_paths=trial_paths, + task_env_config=task_env_config, + image=image, + image_pull_secrets=pull_secrets, + **kwargs, + ) + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + async def start(self, force_build: bool) -> None: + """Ensure the task image exists in the registry, then start the Pod.""" + await self._ensure_client() + await self._ensure_image(force_build=force_build) + await super().start(force_build=False) + + async def stop(self, delete: bool = True) -> None: + """Download proxy data from the Pod, then delete it.""" + if self.proxy_data_dir is not None and self._core_api is not None: + try: + self.logger.debug(f"Downloading proxy data from Pod {self.pod_name}:/proxy-data") + await self.download_dir("/proxy-data", self.proxy_data_dir) + except Exception as exc: + self.logger.warning(f"Failed to download proxy data from Pod {self.pod_name}: {exc}") + await super().stop(delete=delete) + + # ------------------------------------------------------------------ + # Pod spec overrides + # ------------------------------------------------------------------ + + def _build_labels(self) -> dict[str, str]: + labels = super()._build_labels() + labels["ridges.ai/phase"] = "agent" + labels["ridges.ai/evaluation-run-id"] = self.evaluation_run_id + return labels + + def _build_volumes(self) -> list[k8s_client.V1Volume]: + volumes = super()._build_volumes() + volumes.append(k8s_client.V1Volume(name="proxy-certs", empty_dir=k8s_client.V1EmptyDirVolumeSource())) + volumes.append(k8s_client.V1Volume(name="proxy-data", empty_dir=k8s_client.V1EmptyDirVolumeSource())) + return volumes + + def _build_containers(self) -> list[k8s_client.V1Container]: + containers = super()._build_containers() + + # Point the main container at the CA bundle written by the proxy entrypoint. + # ca-bundle.crt = system CAs + proxy CA, so agents trust both external + # HTTPS endpoints and the proxy's self-signed cert. + main = containers[0] + main_env = list(main.env or []) + main_env.append(k8s_client.V1EnvVar(name="SSL_CERT_FILE", value="/proxy-certs/ca-bundle.crt")) + main_env.append(k8s_client.V1EnvVar(name="REQUESTS_CA_BUNDLE", value="/proxy-certs/ca-bundle.crt")) + main.env = main_env + + mounts = list(main.volume_mounts or []) + mounts.append(k8s_client.V1VolumeMount(name="proxy-certs", mount_path="/proxy-certs", read_only=True)) + mounts.append(k8s_client.V1VolumeMount(name="proxy-data", mount_path="/proxy-data", read_only=True)) + main.volume_mounts = mounts + + # Drop all Linux capabilities except SETUID/SETGID which Harbor's + # exec path needs (`su -s /bin/bash -c ...` for exec_as_root + # and exec_as_agent). allowPrivilegeEscalation must remain True for + # the same reason: the kernel requires it for setuid binaries like su. + # runAsNonRoot and readOnlyRootFilesystem are intentionally omitted: + # SWE-bench images build as root (no USER directive), verifiers write to + # /etc and /root at runtime, and conda envs live in root-owned paths. + main.security_context = k8s_client.V1SecurityContext( + capabilities=k8s_client.V1Capabilities( + drop=["ALL"], + add=["SETUID", "SETGID"], + ), + ) + + containers.append(self._proxy_container()) + return containers + + def _build_pod_spec(self) -> k8s_client.V1PodSpec: + spec = super()._build_pod_spec() + + # Prevent untrusted agent code from querying or mutating the K8s API + # via the auto-mounted ServiceAccount token. + spec.automount_service_account_token = False + + # iptables init container: redirect all outbound TCP 443 traffic to the + # proxy sidecar (localhost:15443), except traffic from UID 1337 (the proxy + # itself). This replaces the old hostAliases + socket.getaddrinfo patch + # approach with the standard Istio-style transparent interception pattern. + # Private CIDRs are excluded so in-cluster services are not intercepted. + spec.init_containers = [ + k8s_client.V1Container( + name="iptables-init", + image="alpine:3.20", + image_pull_policy="IfNotPresent", + command=[ + "sh", + "-c", + " && ".join( + [ + "apk add --no-cache iptables", + "iptables -t nat -N PROXY_OUTPUT", + "iptables -t nat -A PROXY_OUTPUT -m owner --uid-owner 1337 -j RETURN", + "iptables -t nat -A PROXY_OUTPUT -d 127.0.0.0/8 -j RETURN", + "iptables -t nat -A PROXY_OUTPUT -d 10.0.0.0/8 -j RETURN", + "iptables -t nat -A PROXY_OUTPUT -d 172.16.0.0/12 -j RETURN", + "iptables -t nat -A PROXY_OUTPUT -d 192.168.0.0/16 -j RETURN", + "iptables -t nat -A PROXY_OUTPUT -p tcp --dport 443 -j REDIRECT --to-ports 15443", + "iptables -t nat -A OUTPUT -p tcp -j PROXY_OUTPUT", + ] + ), + ], + security_context=k8s_client.V1SecurityContext( + capabilities=k8s_client.V1Capabilities(add=["NET_ADMIN"]), + run_as_user=0, + ), + ) + ] + return spec + + def _proxy_container(self) -> k8s_client.V1Container: + env = [ + k8s_client.V1EnvVar(name="MAX_COST_USD", value=self.max_cost_usd), + k8s_client.V1EnvVar(name="EVALUATION_RUN_ID", value=self.evaluation_run_id), + ] + for k, v in self.openrouter_sidecar_env.items(): + # sidecar_env_vars() returns RIDGES_-prefixed names for Docker + # Compose compat; the proxy reads the unprefixed names directly. + name = k.removeprefix("RIDGES_") + env.append(k8s_client.V1EnvVar(name=name, value=v)) + + return k8s_client.V1Container( + name="proxy", + image=self.proxy_image, + # IfNotPresent avoids the `:latest`-tag default of Always, so a + # kind-loaded image (or a cached node image) is used without re-pull. + image_pull_policy="IfNotPresent", + env=env, + # UID 1337 matches the proxy user in the Dockerfile. The iptables + # init container exempts UID 1337 from the port-443 REDIRECT rule, + # so the proxy's own upstream connections reach the internet directly + # while all other containers' port-443 traffic is transparently + # intercepted. + security_context=k8s_client.V1SecurityContext( + run_as_user=1337, + allow_privilege_escalation=False, + capabilities=k8s_client.V1Capabilities(drop=["ALL"]), + ), + volume_mounts=[ + # Mount at /certs/output, not /certs, so the emptyDir does not + # shadow the certs baked into the image at /certs/{ca,server}.* + # The entrypoint writes ca-bundle.crt and ridges-ca.crt here. + k8s_client.V1VolumeMount(name="proxy-certs", mount_path="/certs/output"), + k8s_client.V1VolumeMount(name="proxy-data", mount_path="/proxy-data"), + ], + ports=[k8s_client.V1ContainerPort(container_port=15443)], + ) + + # ------------------------------------------------------------------ + # On-demand image building via Kaniko + # ------------------------------------------------------------------ + + async def _ensure_image(self, *, force_build: bool = False) -> None: + """Check registry for the task image; build with Kaniko if missing.""" + image_ref = f"{self.registry}/{self.task_name}:{self.digest_tag}" + + if not force_build and await self._image_exists_in_registry(image_ref): + self.logger.debug(f"Image {image_ref} already in registry – skipping build") + return + + job_name = f"build-{self._slug(self.task_name)}-{self.digest_tag}" + secret_name = f"{job_name}-url" + self.logger.info(f"Building image {image_ref} via Kaniko job {job_name}") + + try: + await asyncio.to_thread(self._create_build_secret_sync, secret_name) + await asyncio.to_thread(self._create_kaniko_job_sync, job_name, secret_name, image_ref) + except ApiException as exc: + if exc.status == 409: + if await self._is_job_failed(job_name): + self.logger.warning(f"Kaniko job {job_name} previously failed — deleting and retrying") + await self._delete_job(job_name) + await self._delete_secret(secret_name) + await asyncio.to_thread(self._create_build_secret_sync, secret_name) + await asyncio.to_thread(self._create_kaniko_job_sync, job_name, secret_name, image_ref) + else: + self.logger.debug(f"Kaniko job {job_name} already exists — another screener is building") + else: + raise + + await self._wait_for_build_job(job_name, secret_name, timeout_sec=600) + + async def _image_exists_in_registry(self, image_ref: str) -> bool: + """HEAD-check the task image (self.task_name:self.digest_tag) in the registry.""" + import base64 + import http.client + import ssl + import urllib.parse + + name, tag = self.task_name, self.digest_tag + try: + scheme = "http" if self._registry_insecure else "https" + default_port = 5000 if self._registry_insecure else 443 + parsed = urllib.parse.urlsplit(f"{scheme}://{self.registry}") + host = parsed.hostname or self.registry.split(":")[0] + port = parsed.port or default_port + + def _head() -> bool: + if self._registry_insecure: + conn: http.client.HTTPConnection = http.client.HTTPConnection(host, port, timeout=10) + else: + ctx = ssl.create_default_context() + conn = http.client.HTTPSConnection(host, port, timeout=10, context=ctx) + headers: dict[str, str] = {} + if self.registry_credentials_secret and self._registry_password: + cred = base64.b64encode(b"kaniko:" + self._registry_password.encode()).decode() + headers["Authorization"] = f"Basic {cred}" + conn.request("HEAD", f"/v2/{name}/manifests/{tag}", headers=headers) + resp = conn.getresponse() + conn.close() + return resp.status == 200 + + return await asyncio.to_thread(_head) + except Exception as exc: + self.logger.debug(f"Registry check failed for {name}:{tag}: {exc}") + return False + + def _create_build_secret_sync(self, secret_name: str) -> None: + """Create a Secret holding the presigned URL for the Kaniko init container.""" + secret = k8s_client.V1Secret( + metadata=k8s_client.V1ObjectMeta( + name=secret_name, + namespace=self.namespace, + labels={"ridges.ai/managed-by": "screener", "ridges.ai/build-url": "true"}, + ), + string_data={"url": self.task_archive_presigned_url}, + ) + self._api.create_namespaced_secret(namespace=self.namespace, body=secret) + + def _create_kaniko_job_sync(self, job_name: str, secret_name: str, image_ref: str) -> None: + """Create the Kaniko build Job.""" + init_container = k8s_client.V1Container( + name="fetch-context", + image="curlimages/curl:latest", + command=["/bin/sh", "-c"], + args=[ + 'curl -sSfL "$PRESIGNED_URL" -o /tmp/task.tar.gz && ' + "mkdir -p /workspace && " + "tar -xzf /tmp/task.tar.gz -C /workspace --strip-components=1 && " + "test -f /workspace/environment/Dockerfile" + ], + env=[ + k8s_client.V1EnvVar( + name="PRESIGNED_URL", + value_from=k8s_client.V1EnvVarSource( + secret_key_ref=k8s_client.V1SecretKeySelector( + name=secret_name, + key="url", + ) + ), + ) + ], + volume_mounts=[k8s_client.V1VolumeMount(name="context", mount_path="/workspace")], + ) + + kaniko_args = [ + "--dockerfile=Dockerfile", + "--context=dir:///workspace/environment", + f"--destination={image_ref}", + "--cache=true", + f"--cache-repo={self.registry}/cache", + f"--registry-mirror={self.registry}", + ] + if self._registry_insecure: + kaniko_args.append(f"--insecure-registry={self.registry}") + + kaniko_volume_mounts = [ + k8s_client.V1VolumeMount(name="context", mount_path="/workspace"), + ] + if self.registry_credentials_secret: + kaniko_volume_mounts.append( + k8s_client.V1VolumeMount( + name="docker-config", + mount_path="/kaniko/.docker", + read_only=True, + ) + ) + + kaniko_container = k8s_client.V1Container( + name="kaniko", + image="gcr.io/kaniko-project/executor:latest", + args=kaniko_args, + volume_mounts=kaniko_volume_mounts, + ) + + volumes = [ + k8s_client.V1Volume( + name="context", + empty_dir=k8s_client.V1EmptyDirVolumeSource(size_limit="200Mi"), + ) + ] + if self.registry_credentials_secret: + volumes.append( + k8s_client.V1Volume( + name="docker-config", + secret=k8s_client.V1SecretVolumeSource( + secret_name=self.registry_credentials_secret, + items=[ + k8s_client.V1KeyToPath( + key=".dockerconfigjson", + path="config.json", + ) + ], + ), + ) + ) + + job = k8s_client.V1Job( + api_version="batch/v1", + kind="Job", + metadata=k8s_client.V1ObjectMeta( + name=job_name, + namespace=self.namespace, + labels={"ridges.ai/managed-by": "screener", "ridges.ai/build-job": "true"}, + ), + spec=k8s_client.V1JobSpec( + backoff_limit=2, + ttl_seconds_after_finished=300, + template=k8s_client.V1PodTemplateSpec( + metadata=k8s_client.V1ObjectMeta( + labels={"ridges.ai/managed-by": "screener", "ridges.ai/build-job": "true"}, + ), + spec=k8s_client.V1PodSpec( + init_containers=[init_container], + containers=[kaniko_container], + restart_policy="Never", + volumes=volumes, + ), + ), + ), + ) + self._batch.create_namespaced_job(namespace=self.namespace, body=job) + + async def _wait_for_build_job(self, job_name: str, secret_name: str, timeout_sec: int = 600) -> None: + """Poll until the Kaniko Job succeeds or fails.""" + self.logger.debug(f"Waiting for Kaniko build job {job_name} (timeout={timeout_sec}s)") + deadline = asyncio.get_event_loop().time() + timeout_sec + + while asyncio.get_event_loop().time() < deadline: + try: + job = await asyncio.to_thread( + self._batch.read_namespaced_job, + name=job_name, + namespace=self.namespace, + ) + except ApiException as exc: + if exc.status == 404: + await asyncio.sleep(5) + continue + raise + + conditions = job.status.conditions or [] + for cond in conditions: + if cond.type == "Complete" and cond.status == "True": + self.logger.info(f"Kaniko build job {job_name} completed successfully") + # Clean up the secret (best-effort) + await self._delete_secret(secret_name) + return + if cond.type == "Failed" and cond.status == "True": + await self._delete_secret(secret_name) + raise RuntimeError(f"Kaniko build job {job_name} failed: {cond.message}") + + await asyncio.sleep(5) + + await self._delete_secret(secret_name) + raise RuntimeError(f"Kaniko build job {job_name} did not complete within {timeout_sec}s") + + async def _is_job_failed(self, job_name: str) -> bool: + """Return True if the named Job exists and is in Failed state.""" + try: + job = await asyncio.to_thread( + self._batch.read_namespaced_job, + name=job_name, + namespace=self.namespace, + ) + for cond in job.status.conditions or []: + if cond.type == "Failed" and cond.status == "True": + return True + return False + except ApiException: + return False + + async def _delete_job(self, job_name: str) -> None: + """Delete a Kaniko Job (background propagation so pods are also removed).""" + try: + await asyncio.to_thread( + self._batch.delete_namespaced_job, + name=job_name, + namespace=self.namespace, + body=k8s_client.V1DeleteOptions(propagation_policy="Background"), + ) + await asyncio.sleep(2) + except ApiException as exc: + if exc.status != 404: + self.logger.warning(f"Failed to delete job {job_name}: {exc}") + + async def _delete_secret(self, secret_name: str) -> None: + try: + await asyncio.to_thread( + self._api.delete_namespaced_secret, + name=secret_name, + namespace=self.namespace, + ) + except ApiException as exc: + if exc.status != 404: + self.logger.warning(f"Failed to delete build Secret {secret_name}: {exc}") + + @staticmethod + def _slug(name: str) -> str: + """Sanitise a task name for use in Kubernetes resource names.""" + slug = re.sub(r"[^a-z0-9-]", "-", name.lower()) + return slug[:40].strip("-") diff --git a/ridges_harbor/k8s_runtime.py b/ridges_harbor/k8s_runtime.py new file mode 100644 index 00000000..bdf9afc4 --- /dev/null +++ b/ridges_harbor/k8s_runtime.py @@ -0,0 +1,77 @@ +"""Kubernetes-specific runtime helpers for the Ridges proxy-sidecar scaffold. + +Replaces the Docker network-connect logic in ``docker_runtime.py`` for the +Kubernetes execution backend. The single responsibility here is flipping the +``ridges.ai/phase`` Pod label from ``agent`` to ``verification`` so the +namespace-level NetworkPolicy allows full egress during the verifier phase, +and signalling the SNI router inside the proxy container to unlock passthrough. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Any, Awaitable, Callable + +from kubernetes import client as k8s_client + +logger = logging.getLogger(__name__) + +TrialHook = Callable[[Any], Awaitable[None]] + + +def build_k8s_verifier_egress_hook( + *, + namespace: str, + core_api: k8s_client.CoreV1Api, +) -> TrialHook: + """Return a Harbor ``on_verification_started`` hook that enables full egress. + + Two things happen when the hook fires: + + 1. The Pod label ``ridges.ai/phase`` is patched from ``"agent"`` to + ``"verification"``. Combined with the namespace-level NetworkPolicies + (``ridges-agent-egress`` restricts to port 443; ``ridges-verification-egress`` + allows all egress), this gives the verifier unrestricted outbound access. + + 2. ``touch /tmp/egress-unlocked`` is exec'd into the ``proxy`` container so + the SNI router switches from allowlist-only mode to full passthrough. + + The pod name is derived from ``event.trial_id`` at invocation time because + Harbor auto-generates the trial name (and thus the pod name via session_id) + independently of the job name used by the runner. + """ + + async def enable_verifier_egress(event: Any) -> None: + pod_name = event.trial_id.lower().replace("_", "-")[:63] + + # 1. Flip the NetworkPolicy label. + await asyncio.to_thread( + core_api.patch_namespaced_pod, + name=pod_name, + namespace=namespace, + body={"metadata": {"labels": {"ridges.ai/phase": "verification"}}}, + ) + + # 2. Signal the SNI router to unlock passthrough for all egress. + try: + from kubernetes.stream import stream as k8s_stream + + await asyncio.to_thread( + k8s_stream, + core_api.connect_get_namespaced_pod_exec, + name=pod_name, + namespace=namespace, + container="proxy", + command=["touch", "/tmp/egress-unlocked"], + stderr=True, + stdout=True, + stdin=False, + tty=False, + ) + except Exception as exc: + # Non-fatal: the NetworkPolicy label flip already grants egress at + # the network layer. Log and continue so verification is not blocked. + logger.warning("Failed to touch egress-unlocked sentinel in pod %s: %s", pod_name, exc) + + return enable_verifier_egress diff --git a/ridges_harbor/runner.py b/ridges_harbor/runner.py index ac945711..7f417430 100644 --- a/ridges_harbor/runner.py +++ b/ridges_harbor/runner.py @@ -3,7 +3,9 @@ from __future__ import annotations import asyncio +import os import traceback +from collections.abc import Awaitable, Callable from pathlib import Path from typing import Any from uuid import uuid4 @@ -73,6 +75,7 @@ async def run_task( job_name: str | None = None, openrouter_config: OpenRouterRuntimeConfig | None = None, max_cost_usd: float | None = None, + fetch_task_url: Callable[[str], Awaitable[str]] | None = None, inference_seed: int | None = None, on_agent_started: TrialHook | None = None, on_verification_started: TrialHook | None = None, @@ -99,6 +102,7 @@ async def run_task( summary = await _run_task_dir( task_dir=resolved_task_dir, task_name=task_name, + task_digest=task_digest, evaluation_run_id=evaluation_run_id, agent_path=resolved_agent_path, agent_timeout_sec=agent_timeout_sec, @@ -110,6 +114,7 @@ async def run_task( job_name=job_name, openrouter_config=openrouter_config, max_cost_usd=max_cost_usd, + fetch_task_url=fetch_task_url, inference_seed=inference_seed, on_agent_started=on_agent_started, on_verification_started=on_verification_started, @@ -122,6 +127,7 @@ async def _run_task_dir( *, task_dir: Path, task_name: str, + task_digest: str = "", evaluation_run_id: str, agent_path: Path, agent_timeout_sec: float | None, @@ -134,6 +140,7 @@ async def _run_task_dir( openrouter_config: OpenRouterRuntimeConfig | None = None, max_cost_usd: float | None = None, inference_seed: int | None = None, + fetch_task_url: Callable[[str], Awaitable[str]] | None = None, on_agent_started: TrialHook | None = None, on_verification_started: TrialHook | None = None, ) -> HarborRunSummary: @@ -146,6 +153,8 @@ async def _run_task_dir( from harbor.models.job.config import JobConfig, RetryConfig from harbor.models.trial.config import AgentConfig, EnvironmentConfig, TaskConfig, VerifierConfig + ridges_environment_type = os.getenv("RIDGES_ENVIRONMENT_TYPE", "docker") + resolved_job_name = job_name or f"{task_name}__{uuid4().hex[:8]}" job_dir = results_dir / resolved_job_name effective_timeout = agent_timeout_sec if agent_timeout_sec is not None and agent_timeout_sec > 0 else None @@ -167,19 +176,88 @@ async def _run_task_dir( openrouter_config=openrouter_config, ) - environment_config = EnvironmentConfig( - env=docker_environment_env( - ridges_trial_id=ridges_trial_id, - upstream_url=upstream_url, - upstream_host=upstream_host, - evaluation_run_id=evaluation_run_id, - max_cost_usd=effective_max_cost_usd, - proxy_data_dir=str(proxy_data_dir), - openrouter_config=openrouter_config, - inference_seed=inference_seed, + if ridges_environment_type == "kubernetes": + # K8s proxy listens on 8080 (non-root can't bind to 80). + agent_env["SANDBOX_PROXY_URL"] = "http://sandbox-proxy:8080" + + from kubernetes import client as k8s_client_mod + from kubernetes import config as k8s_config_mod + + from validator.config import ( + K8S_CONTEXT, + K8S_NAMESPACE, + K8S_NODE_SELECTOR, + K8S_REGISTRY, + K8S_REGISTRY_INSECURE, + K8S_REGISTRY_PASSWORD, + K8S_REGISTRY_SECRET, + PROXY_IMAGE, ) - ) - config = JobConfig( + + K8S_OWNER_POD_NAME = os.getenv("MY_POD_NAME") + K8S_OWNER_POD_UID = os.getenv("MY_POD_UID") + + from ridges_harbor.k8s_runtime import build_k8s_verifier_egress_hook + + digest_tag = task_digest.split(":")[1][:12] + + # Generate a fresh presigned URL for the Kaniko init container (5-min TTL). + if fetch_task_url is None: + raise RuntimeError("fetch_task_url callback is required in Kubernetes mode") + presigned_url = await fetch_task_url(task_digest) + + environment_config = EnvironmentConfig( + import_path="ridges_harbor.k8s_environment:RidgesKubernetesEnvironment", + env={}, + kwargs={ + "namespace": K8S_NAMESPACE, + "registry": K8S_REGISTRY, + "task_name": task_name, + "digest_tag": digest_tag, + "task_archive_presigned_url": presigned_url, + "proxy_image": PROXY_IMAGE, + "evaluation_run_id": evaluation_run_id, + "max_cost_usd": str(max_cost_usd) if max_cost_usd is not None else "999999", + "openrouter_sidecar_env": openrouter_config.sidecar_env_vars() if openrouter_config else {}, + "proxy_data_dir": str(proxy_data_dir), + "kubeconfig_context": K8S_CONTEXT, + "node_selector": K8S_NODE_SELECTOR, + "labels": {"ridges.ai/trial-id": ridges_trial_id}, + "registry_credentials_secret": K8S_REGISTRY_SECRET, + "registry_password": K8S_REGISTRY_PASSWORD, + "registry_insecure": K8S_REGISTRY_INSECURE, + "owner_pod_name": K8S_OWNER_POD_NAME, + "owner_pod_uid": K8S_OWNER_POD_UID, + }, + ) + + # Build k8s client for the egress hook + try: + k8s_config_mod.load_incluster_config() + except k8s_config_mod.ConfigException: + k8s_config_mod.load_kube_config(context=K8S_CONTEXT) + core_api = k8s_client_mod.CoreV1Api() + + enable_verifier_egress = build_k8s_verifier_egress_hook( + namespace=K8S_NAMESPACE, + core_api=core_api, + ) + else: + environment_config = EnvironmentConfig( + env=docker_environment_env( + ridges_trial_id=ridges_trial_id, + upstream_url=upstream_url, + upstream_host=upstream_host, + evaluation_run_id=evaluation_run_id, + max_cost_usd=effective_max_cost_usd, + proxy_data_dir=str(proxy_data_dir), + openrouter_config=openrouter_config, + inference_seed=inference_seed, + ) + ) + enable_verifier_egress = build_enable_verifier_egress_hook(ridges_trial_id=ridges_trial_id) + + job_config = JobConfig( job_name=resolved_job_name, jobs_dir=results_dir, n_attempts=1, @@ -199,14 +277,13 @@ async def _run_task_dir( ) ], ) - enable_verifier_egress = build_enable_verifier_egress_hook(ridges_trial_id=ridges_trial_id) try: EnvironmentFactory.run_preflight( - type=config.environment.type, - import_path=config.environment.import_path, + type=job_config.environment.type, + import_path=job_config.environment.import_path, ) - job = await Job.create(config) + job = await Job.create(job_config) if on_agent_started is not None: job.on_agent_started(on_agent_started) diff --git a/utils/git.py b/utils/git.py index c57fa47c..a2502ab4 100644 --- a/utils/git.py +++ b/utils/git.py @@ -333,4 +333,32 @@ def get_local_repo_commit_hash(local_repo_dir: str) -> str: ).stdout.strip() -COMMIT_HASH = get_local_repo_commit_hash(pathlib.Path(__file__).parent.parent) +def _read_git_head(repo_dir: pathlib.Path) -> str: + """Resolve HEAD to a commit SHA by reading .git/HEAD directly (no git binary needed).""" + head_file = repo_dir / ".git" / "HEAD" + content = head_file.read_text().strip() + if content.startswith("ref: "): + ref_path = repo_dir / ".git" / content[5:] + return ref_path.read_text().strip() + return content + + +# Resolution order: +# 1. GIT_COMMIT env var (baked in at CI build time as the real SHA) +# 2. git rev-parse HEAD (local dev outside Docker, where git is available) +# 3. Pure-Python .git/HEAD reader (local dev inside Docker container where the +# .git dir is volume-mounted but the git binary is not installed) +_git_commit_env = os.environ.get("GIT_COMMIT", "") + + +def _resolve_commit_hash() -> str: + repo_dir = pathlib.Path(__file__).parent.parent + if _git_commit_env not in ("", "unknown"): + return _git_commit_env + try: + return get_local_repo_commit_hash(repo_dir) + except (FileNotFoundError, subprocess.CalledProcessError): + return _read_git_head(repo_dir) + + +COMMIT_HASH = _resolve_commit_hash() diff --git a/utils/k8s.py b/utils/k8s.py new file mode 100644 index 00000000..df9eeb6f --- /dev/null +++ b/utils/k8s.py @@ -0,0 +1,126 @@ +"""Kubernetes utility helpers — parallel to utils/docker.py.""" + +from __future__ import annotations + +import logging +import os +from typing import Optional + +from kubernetes import client as k8s_client +from kubernetes import config as k8s_config +from kubernetes.client.rest import ApiException + +logger = logging.getLogger(__name__) + +_core_api: Optional[k8s_client.CoreV1Api] = None + + +def _get_core_api() -> k8s_client.CoreV1Api: + global _core_api + if _core_api is None: + try: + k8s_config.load_incluster_config() + except k8s_config.ConfigException: + k8s_config.load_kube_config(context=os.getenv("K8S_CONTEXT")) + _core_api = k8s_client.CoreV1Api() + return _core_api + + +def get_num_k8s_eval_pods() -> int: + """Count running eval Pods (app=ridges-eval) in the namespace.""" + namespace = os.getenv("K8S_NAMESPACE", "ridges") + api = _get_core_api() + pods = api.list_namespaced_pod( + namespace=namespace, + label_selector="app=ridges-eval", + field_selector="status.phase=Running", + ) + return len(pods.items) + + +def cleanup_harbor_k8s_resources() -> None: + """Delete orphaned eval Pods at startup. + + Safe in multi-screener deployments: only deletes pods owned by this + screener (stale UID) or pods with no owner reference (legacy). + """ + namespace = os.getenv("K8S_NAMESPACE", "ridges") + my_pod_name = os.getenv("MY_POD_NAME") + my_pod_uid = os.getenv("MY_POD_UID") + api = _get_core_api() + + logger.info("Cleaning up stale K8s eval Pods...") + + pods = api.list_namespaced_pod( + namespace=namespace, + label_selector="app=ridges-eval", + ) + + removed = 0 + for pod in pods.items: + owner_refs = pod.metadata.owner_references or [] + + if not owner_refs: + # Legacy pod with no owner reference -- always clean up. + pass + elif my_pod_name and my_pod_uid: + # Delete only if an owner ref points to this screener with a stale UID. + dominated_by_us = any(ref.name == my_pod_name and ref.uid != my_pod_uid for ref in owner_refs) + if not dominated_by_us: + continue # Owned by a different screener, or current UID matches. + else: + # No Downward API vars available (local dev?) -- skip pods with owners. + continue + + pod_name = pod.metadata.name + logger.info(f"Removing stale eval Pod {pod_name}...") + try: + api.delete_namespaced_pod( + name=pod_name, + namespace=namespace, + body=k8s_client.V1DeleteOptions( + grace_period_seconds=0, + propagation_policy="Background", + ), + ) + removed += 1 + except ApiException as exc: + if exc.status != 404: + logger.warning(f"Failed to remove stale eval Pod {pod_name}: {exc}") + + logger.info(f"Removed {removed} stale eval Pod(s)") + + +def cleanup_completed_k8s_eval_pods() -> None: + """Delete Succeeded/Failed eval Pods after an evaluation batch. + + Lighter than the startup sweep -- only targets non-Running pods. + """ + namespace = os.getenv("K8S_NAMESPACE", "ridges") + my_pod_name = os.getenv("MY_POD_NAME") + api = _get_core_api() + + for phase in ("Succeeded", "Failed"): + try: + pods = api.list_namespaced_pod( + namespace=namespace, + label_selector="app=ridges-eval", + field_selector=f"status.phase={phase}", + ) + except ApiException: + continue + + for pod in pods.items: + owner_refs = pod.metadata.owner_references or [] + is_ours = not owner_refs or (my_pod_name and any(ref.name == my_pod_name for ref in owner_refs)) + if not is_ours: + continue + + try: + api.delete_namespaced_pod( + name=pod.metadata.name, + namespace=namespace, + body=k8s_client.V1DeleteOptions(grace_period_seconds=0), + ) + except ApiException: + pass diff --git a/utils/system_metrics.py b/utils/system_metrics.py index 7692992c..84ec5121 100644 --- a/utils/system_metrics.py +++ b/utils/system_metrics.py @@ -16,7 +16,7 @@ class SystemMetrics(BaseModel): ram_total_gb: Total RAM in GB disk_percent: Disk percentage (0-100) disk_total_gb: Total disk in GB - num_containers: Number of Docker containers + num_containers: Number of running containers (Docker or k8s eval Pods) """ cpu_percent: Optional[float] = None @@ -41,7 +41,14 @@ async def get_system_metrics() -> SystemMetrics: metrics.disk_percent = disk.percent metrics.disk_total_gb = disk.total / (1000**3) - metrics.num_containers = get_num_docker_containers() + from validator.config import RIDGES_ENVIRONMENT_TYPE + + if RIDGES_ENVIRONMENT_TYPE == "kubernetes": + from utils.k8s import get_num_k8s_eval_pods + + metrics.num_containers = get_num_k8s_eval_pods() + else: + metrics.num_containers = get_num_docker_containers() except Exception as e: logger.warning(f"Error in get_system_metrics(): {e}") diff --git a/validator/config.py b/validator/config.py index ad194759..8acbb891 100644 --- a/validator/config.py +++ b/validator/config.py @@ -223,6 +223,33 @@ logger.info(f"Cleanup Artifact Retention: {CLEANUP_ARTIFACT_RETENTION_HOURS} hour(s)") logger.info(f"Cleanup Task Cache Retention: {CLEANUP_TASK_CACHE_RETENTION_HOURS} hour(s)") -logger.info("Execution Backend: harbor") +# --- Environment Backend --- +RIDGES_ENVIRONMENT_TYPE = os.getenv("RIDGES_ENVIRONMENT_TYPE", "docker") +if RIDGES_ENVIRONMENT_TYPE not in ("docker", "kubernetes"): + logger.fatal("RIDGES_ENVIRONMENT_TYPE must be 'docker' or 'kubernetes'") + +# K8s-only config (only evaluated when RIDGES_ENVIRONMENT_TYPE=kubernetes) +K8S_NAMESPACE: str = "ridges" +K8S_REGISTRY: str = "registry.ridges.svc:5000" +K8S_CONTEXT: str | None = None +K8S_NODE_SELECTOR: dict[str, str] | None = None +K8S_REGISTRY_SECRET: str | None = None +K8S_REGISTRY_PASSWORD: str | None = None +K8S_REGISTRY_INSECURE: bool = True + +PROXY_IMAGE: str = os.getenv("PROXY_IMAGE", "ghcr.io/ridgesai/sandbox-proxy:latest") + +if RIDGES_ENVIRONMENT_TYPE == "kubernetes": + K8S_NAMESPACE = os.getenv("K8S_NAMESPACE", "ridges") + K8S_REGISTRY = os.getenv("K8S_REGISTRY", "registry.ridges.svc:5000") + K8S_CONTEXT = os.getenv("K8S_CONTEXT") # None = use current/in-cluster context + _node_selector_raw = os.getenv("K8S_NODE_SELECTOR") # e.g. "pool=sandbox,arch=amd64" + if _node_selector_raw: + K8S_NODE_SELECTOR = dict(kv.split("=", 1) for kv in _node_selector_raw.split(",")) + K8S_REGISTRY_SECRET = os.getenv("K8S_REGISTRY_SECRET") # e.g. "registry-creds" + K8S_REGISTRY_PASSWORD = os.getenv("K8S_REGISTRY_PASSWORD") # for HEAD check Basic Auth + K8S_REGISTRY_INSECURE = os.getenv("K8S_REGISTRY_INSECURE", "true").lower() == "true" + +logger.info(f"Execution Backend: {RIDGES_ENVIRONMENT_TYPE}") logger.info("===============================") diff --git a/validator/healthz.py b/validator/healthz.py new file mode 100644 index 00000000..a2418b90 --- /dev/null +++ b/validator/healthz.py @@ -0,0 +1,61 @@ +"""Minimal HTTP health endpoint for Kubernetes liveness and readiness probes. + +Usage (in validator/main.py): + + import validator.healthz as healthz + asyncio.create_task(healthz.serve(get_session_id=lambda: session_id)) + +Pod spec: + + livenessProbe: + httpGet: {path: /healthz, port: 8080} + initialDelaySeconds: 30 + periodSeconds: 15 + readinessProbe: + httpGet: {path: /healthz, port: 8080} + initialDelaySeconds: 10 + periodSeconds: 10 +""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Callable +from typing import Any + +logger = logging.getLogger(__name__) + + +async def serve( + *, + get_session_id: Callable[[], Any], + host: str = "0.0.0.0", + port: int = 8080, +) -> None: + """Start the aiohttp health server. Runs forever; call from a background task.""" + try: + from aiohttp import web + except ImportError: + logger.warning("aiohttp not installed – /healthz endpoint disabled (install aiohttp to enable)") + return + + async def healthz(request: web.Request) -> web.Response: + if get_session_id() is not None: + return web.Response(text="ok") + return web.Response(status=503, text="not registered") + + app = web.Application() + app.router.add_get("/healthz", healthz) + + runner = web.AppRunner(app) + await runner.setup() + site = web.TCPSite(runner, host, port) + await site.start() + logger.info(f"Health endpoint listening on {host}:{port}/healthz") + + try: + while True: + await asyncio.sleep(3600) + finally: + await runner.cleanup() diff --git a/validator/main.py b/validator/main.py index 8c66d586..05a7b20b 100644 --- a/validator/main.py +++ b/validator/main.py @@ -4,6 +4,7 @@ import os import pathlib import random +import signal import sys import time import traceback @@ -53,6 +54,8 @@ execution_engine = None STATUS_HOOK_TIMEOUT_SECONDS = 5 +_shutdown_requested = False +_healthz_task: asyncio.Task | None = None # Task-cache digests of in-flight evaluation runs. The cleanup loop excludes these # so it never deletes a cached task a live run is still reading (a task is read for @@ -77,6 +80,32 @@ async def disconnect(reason: str): os._exit(1) +async def _handle_sigterm() -> None: + global _shutdown_requested + logger.warning("SIGTERM received — will shut down after current evaluation finishes") + _shutdown_requested = True + + +async def _run_startup_tasks() -> None: + """Run environment-specific startup tasks before entering the main loop.""" + global _healthz_task + if config.RIDGES_ENVIRONMENT_TYPE == "docker": + cleanup_harbor_docker_resources() + prune_docker_disk_resources(include_build_cache=True) + elif config.RIDGES_ENVIRONMENT_TYPE == "kubernetes": + import validator.healthz as healthz + + _healthz_task = asyncio.create_task(healthz.serve(get_session_id=lambda: session_id)) + from utils.k8s import cleanup_harbor_k8s_resources + + await asyncio.to_thread(cleanup_harbor_k8s_resources) + + asyncio.get_running_loop().add_signal_handler( + signal.SIGTERM, + lambda: asyncio.create_task(_handle_sigterm()), + ) + + # Sends an update-evaluation-run request to the Ridges platform. The extra # parameter is for fields that are not sent in all requests, such as agent_logs # and eval_logs, which are only sent on some state transitions. @@ -648,7 +677,12 @@ async def _run_evaluation(request_evaluation_response: ValidatorRequestEvaluatio ) finally: await _cancel_background_tasks(cancellation_wait_task, poll_task) - await asyncio.to_thread(prune_docker_disk_resources) + if config.RIDGES_ENVIRONMENT_TYPE == "docker": + await asyncio.to_thread(prune_docker_disk_resources) + elif config.RIDGES_ENVIRONMENT_TYPE == "kubernetes": + from utils.k8s import cleanup_completed_k8s_eval_pods + + await asyncio.to_thread(cleanup_completed_k8s_eval_pods) # Main loop @@ -660,8 +694,7 @@ async def main(): global execution_engine setup_logging() - cleanup_harbor_docker_resources() - prune_docker_disk_resources(include_build_cache=True) + await _run_startup_tasks() # Register with the Ridges platform, yielding us a session ID logger.info("Registering validator...") @@ -739,6 +772,11 @@ async def main(): # Loop forever, just keep requesting evaluations and running them while True: + if _shutdown_requested: + logger.info("Shutdown requested — disconnecting gracefully") + await disconnect("SIGTERM (graceful)") + os._exit(0) + logger.info("Requesting an evaluation...") request_evaluation_response_data = await post_ridges_platform(