fix: support PK01 Codex pool sync

2026-07-02 02:43:01 +00:00
parent 18b6b93390
commit 3a8681f458
11 changed files with 258 additions and 48 deletions
@@ -27,12 +27,14 @@ import { resolvedManualAccountProtections } from "./public-exposure";
 import { fieldManager } from "./types";

 export function remotePythonScript(mode: "sync" | "validate" | "trace" | "cleanup-probes" | "sentinel-probe", encodedPayload: string, pool: CodexPoolConfig, target: CodexPoolRuntimeTarget): string {
+  const hostDockerEnvPath = target.runtimeMode === "host-docker" ? target.hostDockerEnvPath : null;
  return `
 set -u
 python3 - <<'PY'
 import base64
 import hashlib
 import json
+import os
 import re
 import secrets
 import string
@@ -43,9 +45,13 @@ from datetime import datetime, timezone, timedelta
 from urllib.parse import quote

 TARGET_ID = ${JSON.stringify(target.id)}
+RUNTIME_MODE = ${JSON.stringify(target.runtimeMode)}
 NAMESPACE = ${JSON.stringify(target.namespace)}
 SERVICE_NAME = ${JSON.stringify(target.serviceName)}
 SERVICE_DNS = ${JSON.stringify(target.serviceDns)}
+HOST_DOCKER_APP_PORT = ${JSON.stringify(target.hostDockerAppPort)}
+HOST_DOCKER_ENV_PATH = ${JSON.stringify(hostDockerEnvPath)}
+HOST_DOCKER_APP_CONTAINER = "sub2api-app"
 FIELD_MANAGER = "${fieldManager}"
 APP_SECRET_NAME = ${JSON.stringify(target.appSecretName)}
 POOL_GROUP_NAME = "${pool.groupName}"
@@ -80,6 +86,107 @@ def text(data, limit=4000):
        data = data.decode("utf-8", errors="replace")
    return data[-limit:]

+def read_host_env():
+    if RUNTIME_MODE != "host-docker":
+        return {}
+    if not isinstance(HOST_DOCKER_ENV_PATH, str) or not HOST_DOCKER_ENV_PATH:
+        raise RuntimeError("host-docker env source path missing")
+    values = {}
+    lines = read_host_env_lines()
+    for line in lines:
+        stripped = line.strip()
+        if not stripped or stripped.startswith("#") or "=" not in stripped:
+            continue
+        key, value = stripped.split("=", 1)
+        key = key.strip()
+        value = value.strip()
+        if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
+            value = value[1:-1]
+        if key:
+            values[key] = value
+    return values
+
+def read_host_env_lines():
+    try:
+        with open(HOST_DOCKER_ENV_PATH, "r", encoding="utf-8") as handle:
+            return handle.read().splitlines()
+    except FileNotFoundError:
+        raise RuntimeError(f"host-docker env source missing: {HOST_DOCKER_ENV_PATH}")
+    except PermissionError:
+        proc = run(["sudo", "-n", "cat", HOST_DOCKER_ENV_PATH])
+        if proc.returncode != 0:
+            raise RuntimeError("read host-docker env source failed: " + text(proc.stderr, 1000))
+        return proc.stdout.decode("utf-8", errors="replace").splitlines()
+
+def write_host_env_value(key, value):
+    if RUNTIME_MODE != "host-docker":
+        raise RuntimeError("write_host_env_value is only valid for host-docker")
+    if not isinstance(HOST_DOCKER_ENV_PATH, str) or not HOST_DOCKER_ENV_PATH:
+        raise RuntimeError("host-docker env source path missing")
+    if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", key):
+        raise RuntimeError(f"unsupported env key: {key}")
+    os.makedirs(os.path.dirname(HOST_DOCKER_ENV_PATH), exist_ok=True)
+    try:
+        lines = read_host_env_lines()
+    except RuntimeError as exc:
+        if "missing" not in str(exc):
+            raise
+        lines = []
+    next_lines = []
+    replaced = False
+    for line in lines:
+        stripped = line.strip()
+        if stripped.startswith("#") or "=" not in stripped:
+            next_lines.append(line)
+            continue
+        current_key = stripped.split("=", 1)[0].strip()
+        if current_key == key:
+            next_lines.append(f"{key}={value}")
+            replaced = True
+        else:
+            next_lines.append(line)
+    if not replaced:
+        next_lines.append(f"{key}={value}")
+    content = "\\n".join(next_lines).rstrip() + "\\n"
+    tmp_path = HOST_DOCKER_ENV_PATH + ".tmp"
+    try:
+        with open(tmp_path, "w", encoding="utf-8") as handle:
+            handle.write(content)
+        os.chmod(tmp_path, 0o600)
+        os.replace(tmp_path, HOST_DOCKER_ENV_PATH)
+    except PermissionError:
+        try:
+            os.unlink(tmp_path)
+        except Exception:
+            pass
+        script = r'''
+set -eu
+path="$1"
+dir="$(dirname "$path")"
+mkdir -p "$dir"
+tmp="$path.tmp.$$"
+umask 077
+cat > "$tmp"
+mv "$tmp" "$path"
+chmod 600 "$path"
+'''
+        proc = run(["sudo", "-n", "sh", "-c", script, "sh", HOST_DOCKER_ENV_PATH], content.encode("utf-8"))
+        if proc.returncode != 0:
+            raise RuntimeError("write host-docker env source failed: " + text(proc.stderr, 1000))
+    return "updated" if replaced else "created"
+
+def docker(args):
+    proc = run(["docker", *args])
+    if proc.returncode == 0:
+        return proc
+    sudo_proc = run(["sudo", "-n", "docker", *args])
+    return sudo_proc if sudo_proc.returncode == 0 else proc
+
+def runtime_logs(since, tail):
+    if RUNTIME_MODE == "host-docker":
+        return docker(["logs", f"--since={since}", f"--tail={tail}", HOST_DOCKER_APP_CONTAINER])
+    return kubectl(["-n", NAMESPACE, "logs", "deployment/sub2api", f"--since={since}", f"--tail={tail}"])
+
 def kubectl(args, input_obj=None):
    if isinstance(input_obj, str):
        input_bytes = input_obj.encode("utf-8")
@@ -98,17 +205,23 @@ def kube_json(args, label):
    return json.loads(raw.decode("utf-8"))

 def decode_secret_value(name, key):
+    if RUNTIME_MODE == "host-docker":
+        return read_host_env().get(key)
    data = kube_json(["-n", NAMESPACE, "get", "secret", name], f"secret/{name}").get("data") or {}
    if key not in data:
        return None
    return base64.b64decode(data[key]).decode("utf-8")

 def get_config_value(name, key):
+    if RUNTIME_MODE == "host-docker":
+        return read_host_env().get(key)
    data = kube_json(["-n", NAMESPACE, "get", "configmap", name], f"configmap/{name}").get("data") or {}
    value = data.get(key)
    return value if isinstance(value, str) and value else None

 def select_app_pod():
+    if RUNTIME_MODE == "host-docker":
+        return HOST_DOCKER_APP_CONTAINER
    pods = kube_json(["-n", NAMESPACE, "get", "pods", "-l", "app.kubernetes.io/name=sub2api"], "sub2api pods").get("items") or []
    for pod in pods:
        status = pod.get("status") or {}
@@ -241,10 +354,15 @@ else
  fi
 fi
 '''
-    proc = run([
-        "kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD,
-        "--", "sh", "-c", script, "sh", method, f"http://127.0.0.1:8080{path}", bearer or "",
-    ], body)
+    if RUNTIME_MODE == "host-docker":
+        if not isinstance(HOST_DOCKER_APP_PORT, int):
+            raise RuntimeError("host-docker app port missing")
+        proc = run(["sh", "-c", script, "sh", method, f"http://127.0.0.1:{HOST_DOCKER_APP_PORT}{path}", bearer or ""], body)
+    else:
+        proc = run([
+            "kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD,
+            "--", "sh", "-c", script, "sh", method, f"http://127.0.0.1:8080{path}", bearer or "",
+        ], body)
    return parse_curl_output(proc)

 def envelope_data(parsed):
@@ -996,6 +1114,9 @@ def ensure_api_key_secret(group_id, token):
        secret_action = "reused-existing-sub2api-key"
    else:
        secret_action = "created"
+    if RUNTIME_MODE == "host-docker":
+        env_action = "kept-existing" if existing else write_host_env_value(POOL_API_KEY_SECRET_KEY, api_key)
+        return api_key, secret_action, f"host-docker-env:{env_action};source={HOST_DOCKER_ENV_PATH};key={POOL_API_KEY_SECRET_KEY};valuesPrinted=false"
    manifest = {
        "apiVersion": "v1",
        "kind": "Secret",
@@ -1022,6 +1143,11 @@ def ensure_api_key_secret(group_id, token):
        raise RuntimeError(f"apply API key secret failed: {text(proc.stderr, 1000)}")
    return api_key, secret_action, text(proc.stdout, 1000)

+def pool_api_key_secret_location():
+    if RUNTIME_MODE == "host-docker":
+        return f"{HOST_DOCKER_ENV_PATH}.{POOL_API_KEY_SECRET_KEY}"
+    return f"{NAMESPACE}/{POOL_API_KEY_SECRET_NAME}.{POOL_API_KEY_SECRET_KEY}"
+
 def apply_sentinel_manifest(manifest):
    if not TARGET_SENTINEL_ENABLED:
        return {
@@ -1190,6 +1316,8 @@ def parse_epoch_z(value):
        return None

 def sentinel_state_object():
+    if not TARGET_SENTINEL_ENABLED:
+        return None, None
    state_name = SENTINEL_CONFIG.get("stateConfigMapName")
    if not state_name:
        return None, None
@@ -1205,6 +1333,8 @@ def sentinel_state_object():
        return obj, None

 def active_sentinel_quarantine_names():
+    if not TARGET_SENTINEL_ENABLED:
+        return set()
    _, state = sentinel_state_object()
    if not isinstance(state, dict):
        return set()
@@ -1668,7 +1798,7 @@ def response_output_preview(parsed):
    return "\\n".join(parts)[:240]

 def request_log_evidence(request_id):
-    proc = kubectl(["-n", NAMESPACE, "logs", "deployment/sub2api", "--since=5m", "--tail=800"])
+    proc = runtime_logs("5m", 800)
    stdout = proc.stdout.decode("utf-8", errors="replace")
    lines = [line for line in stdout.splitlines() if request_id in line]
    failovers = []
@@ -1705,7 +1835,7 @@ def request_log_evidence(request_id):
    }

 def recent_compact_gateway_evidence():
-    proc = kubectl(["-n", NAMESPACE, "logs", "deployment/sub2api", "--since=6h", "--tail=2500"])
+    proc = runtime_logs("6h", 2500)
    stdout = proc.stdout.decode("utf-8", errors="replace")
    failures = []
    successes = []
@@ -1830,7 +1960,7 @@ def failover_budget_exhausted_evidence(failovers, final_errors):
    return exhausted

 def recent_responses_gateway_evidence():
-    proc = kubectl(["-n", NAMESPACE, "logs", "deployment/sub2api", "--since=6h", "--tail=2500"])
+    proc = runtime_logs("6h", 2500)
    stdout = proc.stdout.decode("utf-8", errors="replace")
    failovers = []
    forward_failures = []
@@ -1936,6 +2066,7 @@ def validate_gateway_responses(api_key):
 set -eu
 token="$1"
 request_id="$2"
+url="$3"
 tmp="$(mktemp)"
 trap 'rm -f "$tmp"' EXIT
 cat > "$tmp"
@@ -1945,13 +2076,18 @@ curl -sS -w '\\n__HTTP_CODE__:%{http_code}' -X POST \
  -H "X-Request-ID: $request_id" \
  -H "OpenAI-Client-Request-ID: $request_id" \
  --data-binary @"$tmp" \
-  http://127.0.0.1:8080/v1/responses
+  "$url"
 '''
    started = time.time()
-    proc = run([
-        "kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD,
-        "--", "sh", "-c", script, "sh", api_key, request_id,
-    ], body)
+    if RUNTIME_MODE == "host-docker":
+        if not isinstance(HOST_DOCKER_APP_PORT, int):
+            raise RuntimeError("host-docker app port missing")
+        proc = run(["sh", "-c", script, "sh", api_key, request_id, f"http://127.0.0.1:{HOST_DOCKER_APP_PORT}/v1/responses"], body)
+    else:
+        proc = run([
+            "kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD,
+            "--", "sh", "-c", script, "sh", api_key, request_id, "http://127.0.0.1:8080/v1/responses",
+        ], body)
    resp = parse_curl_output(proc)
    evidence = request_log_evidence(request_id)
    parsed = resp.get("json")
@@ -2123,6 +2259,31 @@ def validate_runtime_capabilities(token):
    }

 def app_pod_runtime_image():
+    if RUNTIME_MODE == "host-docker":
+        proc = docker(["inspect", HOST_DOCKER_APP_CONTAINER])
+        if proc.returncode != 0:
+            return {
+                "container": HOST_DOCKER_APP_CONTAINER,
+                "error": text(proc.stderr, 1000) or text(proc.stdout, 1000),
+            }
+        try:
+            data = json.loads(proc.stdout.decode("utf-8"))
+            item = data[0] if isinstance(data, list) and data else {}
+        except Exception as exc:
+            return {"container": HOST_DOCKER_APP_CONTAINER, "error": str(exc)}
+        state = item.get("State") if isinstance(item, dict) and isinstance(item.get("State"), dict) else {}
+        health = state.get("Health") if isinstance(state.get("Health"), dict) else {}
+        config = item.get("Config") if isinstance(item, dict) and isinstance(item.get("Config"), dict) else {}
+        return {
+            "container": HOST_DOCKER_APP_CONTAINER,
+            "id": (item.get("Id") or "")[:12] if isinstance(item.get("Id"), str) else None,
+            "image": config.get("Image"),
+            "imageID": item.get("Image"),
+            "ready": state.get("Running") is True and (not health or health.get("Status") in (None, "healthy")),
+            "restartCount": item.get("RestartCount"),
+            "startedAt": state.get("StartedAt"),
+            "health": health.get("Status"),
+        }
    try:
        pod = kube_json(["-n", NAMESPACE, "get", "pod", APP_POD], f"pod/{APP_POD}")
        spec_containers = ((pod.get("spec") or {}).get("containers") or []) if isinstance(pod, dict) else []
@@ -2474,7 +2635,7 @@ def run_sync():
        "tempUnschedulable": temp_unschedulable_status,
        "apiKey": {
            "name": POOL_API_KEY_NAME,
-            "secret": f"{NAMESPACE}/{POOL_API_KEY_SECRET_NAME}.{POOL_API_KEY_SECRET_KEY}",
+            "secret": pool_api_key_secret_location(),
            "secretAction": secret_action,
            "secretApply": secret_apply_stdout,
            "sub2apiAction": api_key_result["action"],
@@ -2523,7 +2684,7 @@ def run_validate():
        "appPod": APP_POD,
        "admin": {"email": admin_email, "tokenPrinted": False, "compliance": admin_compliance},
        "apiKey": {
-            "secret": f"{NAMESPACE}/{POOL_API_KEY_SECRET_NAME}.{POOL_API_KEY_SECRET_KEY}",
+            "secret": pool_api_key_secret_location(),
            "sub2apiId": key_item.get("id") if isinstance(key_item, dict) else None,
            "userId": key_item.get("user_id") if isinstance(key_item, dict) else None,
            "groupId": key_item.get("group_id") if isinstance(key_item, dict) else None,