iop/scripts/e2e-openai-managed-capacity-smoke.sh

#!/usr/bin/env bash
set -euo pipefail

# Route-qualified managed OpenAI capacity smoke.
#
# One live invocation owns exactly one public route and one endpoint. It resolves
# that route with the same principal token used at ingress, selects only the
# route-qualified provider snapshot, sends selected capacity + 1 requests, and
# retains raw material only under the ignored agent-test/runs tree. The emitted
# summary is deliberately secret- and payload-free.

SCRIPT_PATH="${BASH_SOURCE[0]}"
SCRIPT_DIR="$(cd "$(dirname "$SCRIPT_PATH")" && pwd -P)"
if [ -d "$SCRIPT_DIR/../agent-test" ]; then
    REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd -P)"
else
    REPO_ROOT="$(pwd -P)"
fi

SELF_TEST=0
MODEL=""
PROVIDER=""
ENDPOINT=""
BASE_URL="https://toki-labs.com:18083/v1"
ROUTES_URL="https://toki-labs.com:18001/v1/credentials/routes"
STATUS_URL="https://toki-labs.com:18001/edges/edge-toki-labs-dev/status"
CA_FILE=""
RUNS_ROOT="$REPO_ROOT/agent-test/runs"
LONG_CONTEXT_THRESHOLD_TOKENS=""
CALLER_TIMEOUT_SECONDS=180
POLL_INTERVAL_SECONDS=0.075
RECOVERY_TIMEOUT_SECONDS=120
RESOLVE_HOST=""
RESOLVE_IP=""
TOKEN_FROM_STDIN=0
RUN_DIR=""
TOKEN=""
AUTH_HEADER_FILE=""
SELF_TEST_TMP_ROOT=""

log() {
    printf '[managed-capacity-smoke] %s\n' "$*"
}

die() {
    log "ERROR: $*" >&2
    exit 1
}

cleanup_live_secret() {
    unset TOKEN 2>/dev/null || true
    TOKEN=""
    if [ -n "${AUTH_HEADER_FILE:-}" ]; then
        rm -f "$AUTH_HEADER_FILE"
        AUTH_HEADER_FILE=""
    fi
}

usage() {
    sed -n '3,43p' "$0"
    cat <<'EOF'

Live usage:
  e2e-openai-managed-capacity-smoke.sh \
    --model <public-route-alias> --provider <resource-selector> \
    --endpoint <chat|responses> --long-context-threshold-tokens <n> \
    --ca-file <ca.pem> --token-stdin [--resolve-host <host> --resolve-ip <ip>]

Self-test:
  e2e-openai-managed-capacity-smoke.sh --self-test

The live token is read as one line from stdin and is never printed. Raw route,
request, response, curl, and status files remain in a new mode-0700 directory
under agent-test/runs. Each invocation handles exactly one route and endpoint.
EOF
}

while [ "$#" -gt 0 ]; do
    case "$1" in
    --self-test)
        SELF_TEST=1
        shift
        ;;
    --model)
        [ "$#" -ge 2 ] || die "--model requires a value"
        MODEL="$2"
        shift 2
        ;;
    --provider)
        [ "$#" -ge 2 ] || die "--provider requires a value"
        PROVIDER="$2"
        shift 2
        ;;
    --endpoint)
        [ "$#" -ge 2 ] || die "--endpoint requires a value"
        ENDPOINT="$2"
        shift 2
        ;;
    --base-url)
        [ "$#" -ge 2 ] || die "--base-url requires a value"
        BASE_URL="${2%/}"
        shift 2
        ;;
    --routes-url)
        [ "$#" -ge 2 ] || die "--routes-url requires a value"
        ROUTES_URL="$2"
        shift 2
        ;;
    --status-url)
        [ "$#" -ge 2 ] || die "--status-url requires a value"
        STATUS_URL="$2"
        shift 2
        ;;
    --ca-file)
        [ "$#" -ge 2 ] || die "--ca-file requires a value"
        CA_FILE="$2"
        shift 2
        ;;
    --runs-root)
        [ "$#" -ge 2 ] || die "--runs-root requires a value"
        RUNS_ROOT="$2"
        shift 2
        ;;
    --long-context-threshold-tokens)
        [ "$#" -ge 2 ] || die "--long-context-threshold-tokens requires a value"
        LONG_CONTEXT_THRESHOLD_TOKENS="$2"
        shift 2
        ;;
    --caller-timeout-seconds)
        [ "$#" -ge 2 ] || die "--caller-timeout-seconds requires a value"
        CALLER_TIMEOUT_SECONDS="$2"
        shift 2
        ;;
    --recovery-timeout-seconds)
        [ "$#" -ge 2 ] || die "--recovery-timeout-seconds requires a value"
        RECOVERY_TIMEOUT_SECONDS="$2"
        shift 2
        ;;
    --resolve-host)
        [ "$#" -ge 2 ] || die "--resolve-host requires a value"
        RESOLVE_HOST="$2"
        shift 2
        ;;
    --resolve-ip)
        [ "$#" -ge 2 ] || die "--resolve-ip requires a value"
        RESOLVE_IP="$2"
        shift 2
        ;;
    --token-stdin)
        TOKEN_FROM_STDIN=1
        shift
        ;;
    -h|--help)
        usage
        exit 0
        ;;
    *)
        die "unknown argument: $1"
        ;;
    esac
done

for required in python3 curl; do
    command -v "$required" >/dev/null 2>&1 || die "$required is required"
done

script_sha256() {
    python3 - "$SCRIPT_PATH" <<'PY'
import hashlib
import pathlib
import sys

print(hashlib.sha256(pathlib.Path(sys.argv[1]).read_bytes()).hexdigest())
PY
}

allocate_run_dir() {
    local root="$1"
    local directory mode
    umask 077
    mkdir -p "$root"
    directory="$(mktemp -d "$root/openai-managed-capacity.XXXXXX")"
    chmod 700 "$directory"
    mode="$(python3 - "$directory" <<'PY'
import os
import stat
import sys

print(oct(stat.S_IMODE(os.stat(sys.argv[1]).st_mode))[2:].zfill(4))
PY
)"
    [ "$mode" = "0700" ] || die "run directory mode is $mode, want 0700"
    printf '%s\n' "$directory"
}

resolve_route_capacity() {
    local routes_file="$1"
    local status_file="$2"
    local model="$3"
    local provider="$4"
    local context_class="$5"
    local claimed_capacity="${6:-}"
    python3 - "$routes_file" "$status_file" "$model" "$provider" "$context_class" "$claimed_capacity" <<'PY'
import json
import sys

routes_path, status_path, model, expected_provider, context_class, claimed = sys.argv[1:]

def field(obj, *names):
    for name in names:
        if name in obj:
            return obj[name]
    return None

routes_doc = json.load(open(routes_path, encoding="utf-8"))
if isinstance(routes_doc, dict):
    routes = field(routes_doc, "routes", "Routes", "data", "Data")
else:
    routes = routes_doc
if not isinstance(routes, list):
    raise SystemExit("route collection is not an array")

matches = []
for route in routes:
    if not isinstance(route, dict):
        continue
    alias = str(field(route, "alias", "Alias") or "").strip()
    status = str(field(route, "status", "Status") or "").strip().lower()
    if alias == model and status == "active":
        matches.append(route)
if len(matches) != 1:
    raise SystemExit(f"active exact route count is {len(matches)}, want 1")

route = matches[0]
selector = str(field(route, "resource_selector", "ResourceSelector") or "").strip()
profile = str(field(route, "profile_id", "ProfileID") or "").strip()
upstream = str(field(route, "upstream_model", "UpstreamModel") or "").strip()
if selector != expected_provider:
    raise SystemExit("route selector does not match expected provider")
if not profile or not upstream:
    raise SystemExit("route profile or upstream model is empty")

status_doc = json.load(open(status_path, encoding="utf-8"))
snapshots = []
for node in status_doc.get("nodes") or status_doc.get("Nodes") or []:
    if not isinstance(node, dict) or field(node, "connected", "Connected") is not True:
        continue
    for snapshot in field(node, "provider_snapshots", "ProviderSnapshots") or []:
        if not isinstance(snapshot, dict):
            continue
        provider_id = str(field(snapshot, "id", "ID", "adapter", "Adapter") or "").strip()
        if provider_id == selector:
            snapshots.append(snapshot)
if len(snapshots) != 1:
    raise SystemExit(f"connected selected-provider snapshot count is {len(snapshots)}, want 1")

snapshot = snapshots[0]
health = str(field(snapshot, "health", "Health") or "").strip().lower()
status = str(field(snapshot, "status", "Status") or "").strip().lower()
if health not in {"healthy", "available"} or status != "available":
    raise SystemExit("selected provider is not healthy and available")
served = field(snapshot, "served_models", "ServedModels") or []
if upstream not in served:
    raise SystemExit("route upstream model is absent from selected provider snapshot")
if context_class == "normal":
    capacity = int(field(snapshot, "capacity", "Capacity") or 0)
elif context_class == "long":
    capacity = int(field(snapshot, "long_context_capacity", "LongContextCapacity") or 0)
else:
    raise SystemExit("context class must be normal or long")
if capacity <= 0:
    raise SystemExit("selected eligible capacity is not positive")
if claimed and int(claimed) != capacity:
    raise SystemExit("claimed capacity includes route-ineligible resources")

print(json.dumps({
    "route_alias": model,
    "selected_provider": selector,
    "profile_present": True,
    "upstream_model_match": True,
    "context_class": context_class,
    "eligible_capacity": capacity,
}, separators=(",", ":"), sort_keys=True))
PY
}

parse_terminal() {
    local endpoint="$1"
    local body_file="$2"
    python3 - "$endpoint" "$body_file" <<'PY'
import json
import pathlib
import sys

endpoint, path = sys.argv[1:]
data = pathlib.Path(path).read_text(encoding="utf-8", errors="strict")
done = finish = completed = errors = 0
data_events = []
for raw_line in data.splitlines():
    if not raw_line.startswith("data:"):
        continue
    payload = raw_line[5:].strip()
    data_events.append(payload)
    if payload == "[DONE]":
        done += 1
        continue
    try:
        obj = json.loads(payload)
    except json.JSONDecodeError as exc:
        raise SystemExit(f"invalid SSE JSON: {exc}")
    if not isinstance(obj, dict):
        raise SystemExit("SSE data payload is not an object")
    event_type = str(obj.get("type") or "")
    if event_type == "response.completed":
        completed += 1
    if event_type in {"error", "run_error", "response.failed", "response.incomplete"} or obj.get("error"):
        errors += 1
    choices = obj.get("choices") or []
    if not isinstance(choices, list):
        raise SystemExit("choices is not an array")
    for choice in choices:
        if isinstance(choice, dict) and choice.get("finish_reason") is not None:
            finish += 1
if not data_events or data_events[-1] != "[DONE]":
    raise SystemExit("[DONE] is missing or is not the final data event")
if done != 1 or errors != 0:
    raise SystemExit(f"terminal counts invalid: done={done} errors={errors}")
if endpoint == "chat":
    if finish != 1 or completed != 0:
        raise SystemExit(f"chat terminal counts invalid: finish={finish} completed={completed}")
elif endpoint == "responses":
    if completed != 1 or finish != 0:
        raise SystemExit(f"responses terminal counts invalid: completed={completed} finish={finish}")
else:
    raise SystemExit("unsupported endpoint")
print(json.dumps({"done_count": done, "finish_count": finish, "completed_count": completed, "error_count": errors}, separators=(",", ":"), sort_keys=True))
PY
}

validate_provenance() {
    local run_dir="$1"
    local manifest="$2"
    python3 - "$run_dir" "$manifest" <<'PY'
import hashlib
import json
import pathlib
import sys

run_dir = pathlib.Path(sys.argv[1]).resolve(strict=True)
manifest_path = pathlib.Path(sys.argv[2]).resolve(strict=True)
if manifest_path.parent != run_dir:
    raise SystemExit("manifest is not owned by the current run directory")
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
run_id = str(manifest.get("run_id") or "")
started_ns = int(manifest.get("run_started_ns") or 0)
dispatch_ns = int(manifest.get("dispatch_started_ns") or 0)
if not run_id or started_ns <= 0 or dispatch_ns <= started_ns:
    raise SystemExit("manifest run identity or timestamps are invalid")
owner = json.loads((run_dir / "run-owner.json").read_text(encoding="utf-8"))
if owner.get("run_id") != run_id or int(owner.get("run_started_ns") or 0) != started_ns:
    raise SystemExit("run owner does not match manifest")
cases = manifest.get("cases")
if not isinstance(cases, list) or not cases:
    raise SystemExit("manifest contains no cases")

seen = set()
for case in cases:
    case_id = str(case.get("case_id") or "")
    if not case_id or case_id in seen:
        raise SystemExit("case identity is empty or duplicated")
    seen.add(case_id)
    for key in ("request", "result", "http_code", "curl_rc", "duration"):
        raw = str(case.get(key) or "")
        path = (run_dir / raw).resolve(strict=True)
        if run_dir not in path.parents:
            raise SystemExit(f"{case_id} {key} is owned by another run")
        stat = path.stat()
        if key == "request":
            if stat.st_mtime_ns < started_ns:
                raise SystemExit(f"{case_id} request predates current run")
            digest = hashlib.sha256(path.read_bytes()).hexdigest()
            if digest != case.get("request_sha256"):
                raise SystemExit(f"{case_id} request digest mismatch")
        else:
            if stat.st_mtime_ns < dispatch_ns:
                raise SystemExit(f"{case_id} {key} predates current dispatch")
    result = (run_dir / str(case["result"])).resolve(strict=True)
    result_digest = hashlib.sha256(result.read_bytes()).hexdigest()
    ownership = json.loads((run_dir / str(case.get("result_owner") or "")).read_text(encoding="utf-8"))
    if ownership.get("run_id") != run_id or ownership.get("case_id") != case_id:
        raise SystemExit(f"{case_id} result owner is foreign")
    if ownership.get("result_sha256") != result_digest:
        raise SystemExit(f"{case_id} result digest mismatch")
print(json.dumps({"run_id": run_id, "case_count": len(cases), "provenance": "current-run"}, separators=(",", ":"), sort_keys=True))
PY
}

write_self_test_manifest() {
    local run_dir="$1"
    local run_id="$2"
    local endpoint="$3"
    local body_file="$4"
    local result_file="$5"
    python3 - "$run_dir" "$run_id" "$endpoint" "$body_file" "$result_file" <<'PY'
import hashlib
import json
import pathlib
import time
import sys

root = pathlib.Path(sys.argv[1]).resolve()
run_id, endpoint = sys.argv[2:4]
request = pathlib.Path(sys.argv[4]).resolve()
result = pathlib.Path(sys.argv[5]).resolve()
started = min(request.stat().st_mtime_ns, result.stat().st_mtime_ns) - 1_000_000
dispatch = result.stat().st_mtime_ns - 1
(root / "run-owner.json").write_text(json.dumps({"run_id": run_id, "run_started_ns": started}), encoding="utf-8")
for name, value in (("http.code", "200\n"), ("curl.rc", "0\n"), ("duration.json", '{"duration_ms":1}\n')):
    (root / name).write_text(value, encoding="utf-8")
owner_path = root / "result-owner.json"
owner_path.write_text(json.dumps({
    "run_id": run_id,
    "case_id": "case-1",
    "result_sha256": hashlib.sha256(result.read_bytes()).hexdigest(),
}), encoding="utf-8")
manifest = {
    "run_id": run_id,
    "run_started_ns": started,
    "dispatch_started_ns": dispatch,
    "endpoint": endpoint,
    "cases": [{
        "case_id": "case-1",
        "request": request.relative_to(root).as_posix(),
        "request_sha256": hashlib.sha256(request.read_bytes()).hexdigest(),
        "result": result.relative_to(root).as_posix(),
        "result_owner": owner_path.relative_to(root).as_posix(),
        "http_code": "http.code",
        "curl_rc": "curl.rc",
        "duration": "duration.json",
    }],
}
(root / "manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")
PY
}

expect_failure() {
    local label="$1"
    shift
    if "$@" >/dev/null 2>&1; then
        die "self-test negative was accepted: $label"
    fi
    log "self-test rejected=$label"
}

run_self_test() {
    local tmp_root run_a run_b mode_a mode_b
    tmp_root="$(mktemp -d "${TMPDIR:-/tmp}/iop-managed-capacity-self-test.XXXXXX")"
    SELF_TEST_TMP_ROOT="$tmp_root"
    trap 'if [ -n "${SELF_TEST_TMP_ROOT:-}" ]; then rm -rf "$SELF_TEST_TMP_ROOT"; fi' EXIT
    run_a="$(allocate_run_dir "$tmp_root")"
    run_b="$(allocate_run_dir "$tmp_root")"
    [ "$run_a" != "$run_b" ] || die "self-test run directories are not unique"
    mode_a="$(python3 -c 'import os,stat,sys; print(oct(stat.S_IMODE(os.stat(sys.argv[1]).st_mode))[2:].zfill(4))' "$run_a")"
    mode_b="$(python3 -c 'import os,stat,sys; print(oct(stat.S_IMODE(os.stat(sys.argv[1]).st_mode))[2:].zfill(4))' "$run_b")"
    [ "$mode_a" = "0700" ] && [ "$mode_b" = "0700" ] || die "self-test run directory mode mismatch"
    log "self-test unique_run_directories=true mode=0700"

    python3 - "$run_a" <<'PY'
import json
import pathlib
import sys

root = pathlib.Path(sys.argv[1])
routes = [{
    "Alias": "ornith:35b", "ProfileID": "openai", "UpstreamModel": "ornith-upstream",
    "ResourceSelector": "onexplayer-lemonade", "Status": "active",
}]
status = {"nodes": [{"connected": True, "provider_snapshots": [
    {"id": "onexplayer-lemonade", "status": "available", "health": "healthy", "capacity": 3,
     "long_context_capacity": 2, "in_flight": 0, "queued": 0, "served_models": ["ornith-upstream"]},
    {"id": "rtx5090-lemonade", "status": "available", "health": "healthy", "capacity": 1,
     "long_context_capacity": 1, "in_flight": 0, "queued": 0, "served_models": ["ornith-upstream"]},
]}]}
(root / "routes.json").write_text(json.dumps(routes), encoding="utf-8")
(root / "status.json").write_text(json.dumps(status), encoding="utf-8")
(root / "request.json").write_text('{"model":"ornith:35b"}', encoding="utf-8")
(root / "chat.sse").write_text('data: {"choices":[{"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n', encoding="utf-8")
(root / "responses.sse").write_text('data: {"type":"response.completed"}\n\ndata: [DONE]\n\n', encoding="utf-8")
(root / "chat-duplicate.sse").write_text('data: {"choices":[{"finish_reason":"stop"}]}\n\ndata: [DONE]\n\ndata: [DONE]\n\n', encoding="utf-8")
(root / "responses-missing.sse").write_text('data: {"type":"response.output_text.done"}\n\ndata: [DONE]\n\n', encoding="utf-8")
PY
    local normal_json long_json
    normal_json="$(resolve_route_capacity "$run_a/routes.json" "$run_a/status.json" "ornith:35b" "onexplayer-lemonade" normal 3)"
    long_json="$(resolve_route_capacity "$run_a/routes.json" "$run_a/status.json" "ornith:35b" "onexplayer-lemonade" long 2)"
    python3 - "$normal_json" "$long_json" <<'PY'
import json
import sys
normal, long = map(json.loads, sys.argv[1:])
assert normal["eligible_capacity"] == 3
assert long["eligible_capacity"] == 2
PY
    expect_failure route-selector-mismatch resolve_route_capacity "$run_a/routes.json" "$run_a/status.json" "ornith:35b" "rtx5090-lemonade" normal
    expect_failure aggregate-capacity-claim resolve_route_capacity "$run_a/routes.json" "$run_a/status.json" "ornith:35b" "onexplayer-lemonade" normal 4
    log "self-test route_exclusion=true normal_capacity=3 long_capacity=2"

    parse_terminal chat "$run_a/chat.sse" >/dev/null
    parse_terminal responses "$run_a/responses.sse" >/dev/null
    expect_failure duplicate-terminal parse_terminal chat "$run_a/chat-duplicate.sse"
    expect_failure missing-terminal parse_terminal responses "$run_a/responses-missing.sse"
    log "self-test exact_chat_and_responses_terminals=true"

    write_self_test_manifest "$run_a" self-test-a chat "$run_a/request.json" "$run_a/chat.sse"
    validate_provenance "$run_a" "$run_a/manifest.json" >/dev/null
    python3 - "$run_a" <<'PY'
import json
import os
import pathlib
import sys
root = pathlib.Path(sys.argv[1])
manifest = json.loads((root / "manifest.json").read_text())
old = manifest["dispatch_started_ns"] - 1_000_000_000
os.utime(root / "chat.sse", ns=(old, old))
PY
    expect_failure stale-body validate_provenance "$run_a" "$run_a/manifest.json"

    python3 - "$run_a" <<'PY'
import json
import pathlib
import sys
root = pathlib.Path(sys.argv[1])
manifest = json.loads((root / "manifest.json").read_text())
manifest["cases"][0]["result"] = "missing.sse"
(root / "manifest-missing.json").write_text(json.dumps(manifest), encoding="utf-8")
PY
    expect_failure missing-body validate_provenance "$run_a" "$run_a/manifest-missing.json"

    python3 - "$run_a" "$run_b" <<'PY'
import json
import pathlib
import sys
root, other = map(pathlib.Path, sys.argv[1:])
(other / "foreign.sse").write_text('data: [DONE]\n\n', encoding="utf-8")
manifest = json.loads((root / "manifest.json").read_text())
manifest["cases"][0]["result"] = str((other / "foreign.sse").resolve())
(root / "manifest-foreign.json").write_text(json.dumps(manifest), encoding="utf-8")
PY
    expect_failure foreign-run-body validate_provenance "$run_a" "$run_a/manifest-foreign.json"
    log "self-test current_manifest_acceptance=true stale_missing_foreign_rejected=true"
    log "SELF_TEST_PASS"
}

if [ "$SELF_TEST" -eq 1 ]; then
    run_self_test
    exit 0
fi

[ -n "$MODEL" ] || die "--model is required"
[ -n "$PROVIDER" ] || die "--provider is required"
case "$ENDPOINT" in chat|responses) ;; *) die "--endpoint must be chat or responses" ;; esac
case "$LONG_CONTEXT_THRESHOLD_TOKENS" in ''|*[!0-9]*) die "--long-context-threshold-tokens must be a positive integer" ;; esac
[ "$LONG_CONTEXT_THRESHOLD_TOKENS" -gt 0 ] || die "--long-context-threshold-tokens must be positive"
case "$CALLER_TIMEOUT_SECONDS" in ''|*[!0-9]*) die "--caller-timeout-seconds must be a positive integer" ;; esac
case "$RECOVERY_TIMEOUT_SECONDS" in ''|*[!0-9]*) die "--recovery-timeout-seconds must be a positive integer" ;; esac
[ "$TOKEN_FROM_STDIN" -eq 1 ] || die "--token-stdin is required for live mode"
[ -n "$CA_FILE" ] && [ -f "$CA_FILE" ] || die "--ca-file must name a readable file"
if { [ -n "$RESOLVE_HOST" ] && [ -z "$RESOLVE_IP" ]; } || { [ -z "$RESOLVE_HOST" ] && [ -n "$RESOLVE_IP" ]; }; then
    die "--resolve-host and --resolve-ip must be supplied together"
fi
IFS= read -r TOKEN || die "failed to read principal token from stdin"
[ -n "$TOKEN" ] || die "principal token is empty"

RUN_DIR="$(allocate_run_dir "$RUNS_ROOT")"
RUN_ID="$(python3 - <<'PY'
import secrets
import time
print(f"{time.time_ns()}-{secrets.token_hex(8)}")
PY
)"
RUN_STARTED_NS="$(python3 -c 'import time; print(time.time_ns())')"
SCRIPT_SHA256="$(script_sha256)"
python3 - "$RUN_DIR/run-owner.json" "$RUN_ID" "$RUN_STARTED_NS" <<'PY'
import json
import pathlib
import sys
path, run_id, started = sys.argv[1:]
pathlib.Path(path).write_text(json.dumps({"run_id": run_id, "run_started_ns": int(started)}), encoding="utf-8")
PY
AUTH_HEADER_FILE="$RUN_DIR/auth-header"
printf 'Authorization: Bearer %s\n' "$TOKEN" >"$AUTH_HEADER_FILE"
chmod 600 "$AUTH_HEADER_FILE"
unset TOKEN
TOKEN=""
trap cleanup_live_secret EXIT
trap 'cleanup_live_secret; exit 1' HUP INT TERM

CURL_COMMON=(--noproxy '*' --connect-timeout 10 --cacert "$CA_FILE")
CURL_RESOLVE=()
if [ -n "$RESOLVE_HOST" ]; then
    while IFS= read -r resolve; do
        [ -n "$resolve" ] && CURL_RESOLVE+=(--resolve "$resolve")
    done < <(python3 - "$RESOLVE_HOST" "$RESOLVE_IP" "$BASE_URL" "$ROUTES_URL" "$STATUS_URL" <<'PY'
import sys
import urllib.parse
host, ip = sys.argv[1:3]
seen = set()
for value in sys.argv[3:]:
    parsed = urllib.parse.urlparse(value)
    port = parsed.port or (443 if parsed.scheme == "https" else 80)
    item = f"{host}:{port}:{ip}"
    if item not in seen:
        print(item)
        seen.add(item)
PY
)
fi

curl "${CURL_COMMON[@]}" ${CURL_RESOLVE[@]+"${CURL_RESOLVE[@]}"} --max-time 15 -fsS \
    -H "@$AUTH_HEADER_FILE" "$ROUTES_URL" >"$RUN_DIR/routes.json"
curl "${CURL_COMMON[@]}" ${CURL_RESOLVE[@]+"${CURL_RESOLVE[@]}"} --max-time 10 -fsS \
    "$STATUS_URL" >"$RUN_DIR/status-initial.json"

REQUEST_INFO="$(python3 - "$RUN_DIR" "$MODEL" "$ENDPOINT" "$LONG_CONTEXT_THRESHOLD_TOKENS" <<'PY'
import hashlib
import json
import pathlib
import sys

root = pathlib.Path(sys.argv[1])
model, endpoint, threshold = sys.argv[2], sys.argv[3], int(sys.argv[4])
prompt = (
    "Write a generic capacity qualification report of approximately 850 tokens. "
    "Use exactly ten numbered sections with short headings and concrete but non-sensitive operational observations. "
    "Do not use tools or hidden reasoning. End with the marker MANAGED_CAPACITY_COMPLETE."
)
common = {
    "model": model,
    "stream": True,
    "temperature": 0,
    "chat_template_kwargs": {"enable_thinking": False},
}
if endpoint == "chat":
    body = dict(common, messages=[{"role": "user", "content": prompt}], max_tokens=1200)
else:
    body = dict(common, input=prompt, max_output_tokens=1200)
encoded = json.dumps(body, ensure_ascii=False, separators=(",", ":"))
runes = len(encoded)
estimate = runes // 4 + runes // 16
context_class = "long" if estimate >= threshold else "normal"
path = root / "request-template.json"
path.write_text(encoded, encoding="utf-8")
print(json.dumps({
    "request_runes": runes,
    "estimated_input_tokens": max(1, estimate),
    "context_class": context_class,
    "request_sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
}, separators=(",", ":"), sort_keys=True))
PY
)"
CONTEXT_CLASS="$(python3 -c 'import json,sys; print(json.loads(sys.argv[1])["context_class"])' "$REQUEST_INFO")"
[ "$CONTEXT_CLASS" = "normal" ] || die "managed normal-capacity smoke emitted a $CONTEXT_CLASS request"
ROUTE_INFO="$(resolve_route_capacity "$RUN_DIR/routes.json" "$RUN_DIR/status-initial.json" "$MODEL" "$PROVIDER" "$CONTEXT_CLASS")"
ELIGIBLE_CAPACITY="$(python3 -c 'import json,sys; print(json.loads(sys.argv[1])["eligible_capacity"])' "$ROUTE_INFO")"
REQUEST_COUNT=$((ELIGIBLE_CAPACITY + 1))

python3 - "$RUN_DIR" "$RUN_ID" "$RUN_STARTED_NS" "$MODEL" "$PROVIDER" "$ENDPOINT" "$REQUEST_COUNT" "$REQUEST_INFO" <<'PY'
import hashlib
import json
import pathlib
import shutil
import sys

root = pathlib.Path(sys.argv[1])
run_id, started, model, provider, endpoint = sys.argv[2:7]
request_count = int(sys.argv[7])
request_info = json.loads(sys.argv[8])
template = root / "request-template.json"
cases = []
for index in range(1, request_count + 1):
    case_id = f"case-{index}"
    case_dir = root / case_id
    case_dir.mkdir(mode=0o700)
    request = case_dir / "request.json"
    shutil.copyfile(template, request)
    cases.append({
        "case_id": case_id,
        "request": request.relative_to(root).as_posix(),
        "request_sha256": hashlib.sha256(request.read_bytes()).hexdigest(),
        "result": f"{case_id}/response.sse",
        "result_owner": f"{case_id}/result-owner.json",
        "http_code": f"{case_id}/http.code",
        "curl_rc": f"{case_id}/curl.rc",
        "duration": f"{case_id}/duration.json",
    })
manifest = {
    "run_id": run_id,
    "run_started_ns": int(started),
    "dispatch_started_ns": 0,
    "route_alias": model,
    "selected_provider": provider,
    "endpoint": endpoint,
    "request_shape": request_info,
    "cases": cases,
}
(root / "manifest.json").write_text(json.dumps(manifest, indent=2, sort_keys=True), encoding="utf-8")
PY

DISPATCH_STARTED_NS="$(python3 -c 'import time; print(time.time_ns())')"
python3 - "$RUN_DIR/manifest.json" "$DISPATCH_STARTED_NS" <<'PY'
import json
import pathlib
import sys
path = pathlib.Path(sys.argv[1])
manifest = json.loads(path.read_text(encoding="utf-8"))
manifest["dispatch_started_ns"] = int(sys.argv[2])
path.write_text(json.dumps(manifest, indent=2, sort_keys=True), encoding="utf-8")
PY

if [ "$ENDPOINT" = "chat" ]; then
    API_URL="$BASE_URL/chat/completions"
else
    API_URL="$BASE_URL/responses"
fi

run_case() {
    local index="$1"
    local case_id="case-$index"
    local case_dir="$RUN_DIR/$case_id"
    local start_ns end_ns duration_ms rc http
    for path in response.sse http.code curl.rc curl.stderr duration.json result-owner.json; do
        [ ! -e "$case_dir/$path" ] || return 91
    done
    # Each timestamp is collected by a separate Python process.  On macOS,
    # monotonic clock epochs are not guaranteed to be comparable across those
    # processes, so use the shared wall-clock epoch for the elapsed-time field.
    start_ns="$(python3 -c 'import time; print(time.time_ns())')"
    set +e
    http="$(curl "${CURL_COMMON[@]}" ${CURL_RESOLVE[@]+"${CURL_RESOLVE[@]}"} --max-time "$CALLER_TIMEOUT_SECONDS" \
        --no-buffer -sS -o "$case_dir/response.sse" -w '%{http_code}' \
        -H "@$AUTH_HEADER_FILE" -H 'Content-Type: application/json' \
        --data-binary "@$case_dir/request.json" "$API_URL" 2>"$case_dir/curl.stderr")"
    rc=$?
    set -e
    end_ns="$(python3 -c 'import time; print(time.time_ns())')"
    duration_ms=$(((end_ns - start_ns) / 1000000))
    printf '%s\n' "$http" >"$case_dir/http.code"
    printf '%s\n' "$rc" >"$case_dir/curl.rc"
    printf '{"duration_ms":%s}\n' "$duration_ms" >"$case_dir/duration.json"
    python3 - "$case_dir/result-owner.json" "$RUN_ID" "$case_id" "$case_dir/response.sse" <<'PY'
import hashlib
import json
import pathlib
import sys
owner, run_id, case_id, result = sys.argv[1:]
pathlib.Path(owner).write_text(json.dumps({
    "run_id": run_id,
    "case_id": case_id,
    "result_sha256": hashlib.sha256(pathlib.Path(result).read_bytes()).hexdigest(),
}), encoding="utf-8")
PY
}

fetch_status() {
    local output="$1"
    curl "${CURL_COMMON[@]}" ${CURL_RESOLVE[@]+"${CURL_RESOLVE[@]}"} --max-time 10 -fsS "$STATUS_URL" >"$output"
}

append_selected_observation() {
    local snapshot="$1"
    python3 - "$snapshot" "$PROVIDER" <<'PY'
import json
import sys

doc = json.load(open(sys.argv[1], encoding="utf-8"))
provider_id = sys.argv[2]
matches = []
for node in doc.get("nodes") or []:
    if node.get("connected") is not True:
        continue
    for snapshot in node.get("provider_snapshots") or []:
        if (snapshot.get("id") or snapshot.get("adapter")) == provider_id:
            matches.append(snapshot)
if len(matches) != 1:
    raise SystemExit("selected provider observation is not unique")
s = matches[0]
print(json.dumps({
    "capacity": int(s.get("capacity") or 0),
    "in_flight": int(s.get("in_flight") or 0),
    "queued": int(s.get("queued") or 0),
    "health": str(s.get("health") or ""),
    "status": str(s.get("status") or ""),
}, separators=(",", ":"), sort_keys=True))
PY
}

pids=()
index=1
while [ "$index" -le "$REQUEST_COUNT" ]; do
    run_case "$index" &
    pids+=("$!")
    index=$((index + 1))
done

sample=0
while :; do
    alive=0
    for pid in "${pids[@]}"; do
        if kill -0 "$pid" 2>/dev/null; then
            alive=1
        fi
    done
    snapshot="$RUN_DIR/status-$sample.json"
    if fetch_status "$snapshot"; then
        append_selected_observation "$snapshot" >>"$RUN_DIR/status-observations.ndjson"
    else
        printf 'status_fetch_failed sample=%s\n' "$sample" >>"$RUN_DIR/status-errors.log"
    fi
    sample=$((sample + 1))
    [ "$alive" -eq 1 ] || break
    sleep "$POLL_INTERVAL_SECONDS"
done
for pid in "${pids[@]}"; do
    if wait "$pid"; then
        :
    else
        :
    fi
done
cleanup_live_secret

recovery_deadline=$((SECONDS + RECOVERY_TIMEOUT_SECONDS))
while :; do
    snapshot="$RUN_DIR/status-recovery-$sample.json"
    if fetch_status "$snapshot"; then
        observation="$(append_selected_observation "$snapshot")"
        printf '%s\n' "$observation" >>"$RUN_DIR/status-observations.ndjson"
        if python3 - "$observation" <<'PY'
import json
import sys
item = json.loads(sys.argv[1])
raise SystemExit(0 if item["in_flight"] == 0 and item["queued"] == 0 and item["health"] in {"healthy", "available"} and item["status"] == "available" else 1)
PY
        then
            break
        fi
    fi
    [ "$SECONDS" -lt "$recovery_deadline" ] || die "selected provider did not recover before the bounded deadline"
    sample=$((sample + 1))
    sleep "$POLL_INTERVAL_SECONDS"
done

validate_provenance "$RUN_DIR" "$RUN_DIR/manifest.json" >"$RUN_DIR/provenance-summary.json"
index=1
while [ "$index" -le "$REQUEST_COUNT" ]; do
    parse_terminal "$ENDPOINT" "$RUN_DIR/case-$index/response.sse" >"$RUN_DIR/case-$index/terminal-summary.json"
    index=$((index + 1))
done

python3 - "$RUN_DIR" "$SCRIPT_SHA256" "$ROUTE_INFO" "$REQUEST_INFO" "$ELIGIBLE_CAPACITY" <<'PY'
import json
import pathlib
import sys

root = pathlib.Path(sys.argv[1])
script_sha = sys.argv[2]
route = json.loads(sys.argv[3])
shape = json.loads(sys.argv[4])
capacity = int(sys.argv[5])
manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
observations = [json.loads(line) for line in (root / "status-observations.ndjson").read_text(encoding="utf-8").splitlines() if line]
if not observations:
    raise SystemExit("no selected-provider observations were captured")
if any(item["capacity"] != capacity for item in observations):
    raise SystemExit("selected provider capacity changed during the invocation")
peak = max(item["in_flight"] for item in observations)
max_queue = max(item["queued"] for item in observations)
if peak != capacity:
    raise SystemExit(f"selected provider peak={peak}, want capacity={capacity}")
if max_queue < 1:
    raise SystemExit("selected provider queue was not observed")
if any(item["in_flight"] > capacity for item in observations):
    raise SystemExit("selected provider exceeded eligible capacity")
final = observations[-1]
if final["in_flight"] != 0 or final["queued"] != 0:
    raise SystemExit("selected provider final counters did not recover")

http_statuses = []
durations = []
done = finish = completed = errors = 0
for case in manifest["cases"]:
    case_root = root / case["case_id"]
    rc = int((case_root / "curl.rc").read_text().strip())
    http = (case_root / "http.code").read_text().strip()
    if rc != 0 or http != "200":
        raise SystemExit(f"{case['case_id']} curl/http failed")
    http_statuses.append(int(http))
    durations.append(int(json.loads((case_root / "duration.json").read_text())["duration_ms"]))
    terminal = json.loads((case_root / "terminal-summary.json").read_text())
    done += terminal["done_count"]
    finish += terminal["finish_count"]
    completed += terminal["completed_count"]
    errors += terminal["error_count"]

summary = {
    "evidence_schema": "iop.openai_managed_capacity_smoke.v1",
    "run_id": manifest["run_id"],
    "script_sha256": script_sha,
    "route_alias": route["route_alias"],
    "selected_provider": route["selected_provider"],
    "endpoint": manifest["endpoint"],
    "request_count": len(manifest["cases"]),
    "request_runes": shape["request_runes"],
    "estimated_input_tokens": shape["estimated_input_tokens"],
    "context_class": shape["context_class"],
    "eligible_capacity": capacity,
    "http_200_count": sum(1 for value in http_statuses if value == 200),
    "duration_ms_min": min(durations),
    "duration_ms_max": max(durations),
    "done_count": done,
    "finish_count": finish,
    "completed_count": completed,
    "error_count": errors,
    "selected_peak_in_flight": peak,
    "selected_max_queued": max_queue,
    "selected_final_in_flight": final["in_flight"],
    "selected_final_queued": final["queued"],
    "provenance": "current-run-manifest",
    "outcome": "pass",
}
(root / "sanitized-summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8")
print(json.dumps(summary, separators=(",", ":"), sort_keys=True))
PY
log "PASS evidence_dir=$RUN_DIR"