scripts/smoke_hf_openoperator.py

#!/usr/bin/env python3
"""Secret-safe, non-destructive OpenOperator Hugging Face acceptance smoke."""

from __future__ import annotations
import json
import os
import argparse
import time
import urllib.error
import urllib.request
import uuid

BASE = os.getenv(
    "OPENOPERATOR_BASE_URL", "https://leon4gr45-openoperator.hf.space"
).rstrip("/")
HF_TOKEN = os.getenv("HF_TOKEN", "").strip()
API_KEY = os.getenv("SPYNEL_AGENT_ZERO_API_KEY", "").strip()
EXPECTED_SHA = os.getenv("EXPECTED_OPENOPERATOR_SHA", "").strip()
TIMEOUT = float(os.getenv("OPENOPERATOR_SMOKE_TIMEOUT", "30"))
POLL = float(os.getenv("SPYNEL_POLL_INTERVAL", "2"))
results = []
REQUIRED_PROFILES = {
    "developer",
    "hacker",
    "spynel",
    "reviewer",
    "tester",
    "tiny-coder",
    "debugger",
    "integrator",
    "frontend-qa",
    "evals",
    "shipper",
    "launch",
    "performance",
    "security",
    "refactorer",
    "maintainer",
    "data-engineer",
    "docs",
    "repository-manager",
    "prompt-improver",
}

CRITICAL_SKILLS = {
    "colab-execute",
    "golive",
    "brag",
    "playwright",
    "hf-space",
    "test-evidence",
    "api-contract",
    "benchmark",
    "compile-check",
    "context-packager",
    "dependency-doctor",
    "diff-self-review",
    "docker-diagnose",
    "failure-to-next-patch",
    "git-worktree",
    "github-pr",
    "patch-small",
    "repo-map",
    "secret-safe-env",
    "spec-check",
    "symbol-locator",
    "task-slicer",
    "test-targeted",
}


def request(path, payload=None, *, api=False):
    headers = {"Accept": "application/json"}
    if HF_TOKEN:
        headers["Authorization"] = "Bearer " + HF_TOKEN
    if api:
        if not API_KEY:
            raise RuntimeError("SPYNEL_AGENT_ZERO_API_KEY is not configured")
        headers["X-API-KEY"] = API_KEY
    data = None
    if payload is not None:
        data = json.dumps(payload).encode()
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(
        BASE + path,
        data=data,
        headers=headers,
        method="POST" if payload is not None else "GET",
    )
    try:
        with urllib.request.urlopen(req, timeout=TIMEOUT) as response:
            raw = response.read().decode(errors="replace")
            return response.status, json.loads(raw) if raw else {}
    except urllib.error.HTTPError as exc:
        return exc.code, {"error": "HTTP error"}


def report(name, state, detail=""):
    results.append((name, state))
    print(f"{name}: {state}" + (f" ({detail})" if detail else ""))


def poll(context_id, *, include_logs=False):
    deadline = time.monotonic() + TIMEOUT
    while time.monotonic() < deadline:
        code, data = request(
            "/api/api_poll",
            {"context_id": context_id, "include_logs": include_logs},
            api=True,
        )
        if code != 200:
            return False, data
        if data.get("requires_attention"):
            return False, data
        if data.get("status") in {"completed", "complete"}:
            if include_logs:
                time.sleep(max(0.25, min(2.0, POLL)))
                final_code, final_data = request(
                    "/api/api_poll",
                    {"context_id": context_id, "include_logs": True},
                    api=True,
                )
                if final_code == 200:
                    data = final_data
            return True, data
        if data.get("status") in {
            "failed",
            "blocked",
            "cancelled",
            "partially_verified",
        }:
            return False, data
        time.sleep(max(0.25, min(POLL, float(data.get("next_poll_after") or POLL))))
    return False, {"error": "finite timeout"}


def _names(value):
    """Extract public catalog names without depending on a single response shape."""
    if isinstance(value, dict):
        value = value.get("data", value.get("skills", value.get("agents", [])))
    if not isinstance(value, list):
        return set()
    return {
        str((item.get("name") or item.get("key")) if isinstance(item, dict) else item)
        .strip()
        .lower()
        for item in value
        if ((item.get("name") or item.get("key")) if isinstance(item, dict) else item)
    }


def discover():
    """Use existing read-only authenticated catalogs; never mutate for discovery."""
    code, agents = request("/api/agents", {"action": "list"}, api=True)
    names = _names(agents)
    missing_profiles = REQUIRED_PROFILES - names
    report(
        "specialist profiles",
        "PASS" if code == 200 and not missing_profiles else "FAIL",
        f"missing: {sorted(missing_profiles)}" if missing_profiles else f"{len(names)} profiles cataloged",
    )
    report("Spynel", "PASS" if "spynel" in names else "FAIL")

    code, skills = request(
        "/api/plugins/_skills/skills_catalog", {"action": "list"}, api=True
    )
    skill_names = _names(skills)
    missing_skills = CRITICAL_SKILLS - skill_names
    report(
        "required skills",
        "PASS" if code == 200 and not missing_skills else "FAIL",
        f"missing: {sorted(missing_skills)}" if missing_skills else f"{len(skill_names)} skills cataloged",
    )

    from helpers import files
    spynel_skills_dir = files.get_abs_path("agents/spynel/skills")
    spynel_skills = set(files.get_subdirectories(spynel_skills_dir)) if files.exists(spynel_skills_dir) else set()
    report("spynel-dispatch", "PASS" if "spynel-dispatch" in spynel_skills or "spynel-dispatch" in skill_names else "FAIL")
    report("paperclip-board-manager", "PASS" if "paperclip-board-manager" in spynel_skills or "paperclip-board-manager" in skill_names else "FAIL")


def _settings_acceptance(data):
    additional = data.get("additional", {})
    configured = data.get("settings", {})
    runtime = data.get("runtime", {})
    model = data.get("main_model", {})
    report("settings", "PASS")
    report(
        "runtime",
        "PASS"
        if additional.get("is_dockerized")
        and not additional.get("is_development", False)
        else "FAIL",
        "dockerized production expected",
    )
    report(
        "non-root",
        "PASS"
        if runtime.get("uid") not in {None, 0} and runtime.get("gid") not in {None, 0}
        else "FAIL",
        "non-root uid/gid expected",
    )
    report(
        "root password capability",
        "PASS" if additional.get("root_password_supported") is False else "FAIL",
        "unsupported expected",
    )
    serialized = json.dumps(data)
    report(
        "API key",
        "PASS"
        if configured.get("mcp_server_token") == "************"
        and "BLABLADOR_API_KEY" not in serialized
        else "FAIL",
        "configured/masked",
    )
    report(
        "compatible provider",
        "PASS"
        if model.get("provider") == "other"
        and model.get("name")
        and model.get("api_base")
        else "FAIL",
    )


def _reversible_settings_smoke(data):
    original = data.get("settings", {}).get("time_format")
    alternate = "24h" if original == "12h" else "12h"
    code, changed = request(
        "/api/settings_probe",
        {"action": "set_time_format", "value": alternate},
        api=True,
    )
    code2, reloaded = request("/api/settings_probe", {"action": "get"}, api=True)
    persisted = (
        code == 200
        and code2 == 200
        and reloaded.get("settings", {}).get("time_format") == alternate
    )
    restore_code, _ = request(
        "/api/settings_probe",
        {"action": "set_time_format", "value": original},
        api=True,
    )
    restore_get, restored = request("/api/settings_probe", {"action": "get"}, api=True)
    restored_ok = (
        restore_code == 200
        and restore_get == 200
        and restored.get("settings", {}).get("time_format") == original
    )
    report(
        "settings persistence",
        "PASS" if persisted and restored_ok else "FAIL",
        "safe setting changed, reloaded, and restored",
    )


def main(mode="full", expected_sha=""):
    target_sha = expected_sha or EXPECTED_SHA
    code, data = request("/health")
    health_pass = code == 200 and data.get("status") == "ok"
    report("health", "PASS" if health_pass else "FAIL")
    if target_sha:
        deployed_sha = str(data.get("sha") or "").strip()
        sha_match = deployed_sha == target_sha
        report("deployed source SHA", "PASS" if sha_match else "FAIL", f"expected {target_sha}, got {deployed_sha}")
    if mode == "basic":
        return 1 if any(state != "PASS" for _, state in results) else 0
    if API_KEY:
        code, data = request("/api/settings_probe", {"action": "get"}, api=True)
        if code == 200:
            _settings_acceptance(data)
            if mode == "full":
                _reversible_settings_smoke(data)
        else:
            report("settings", "FAIL", "authenticated settings probe failed")
    else:
        report("settings", "FAIL", "SPYNEL_AGENT_ZERO_API_KEY absent")
    if not API_KEY:
        report("chat", "BLOCKED", "SPYNEL_AGENT_ZERO_API_KEY absent")
        report("delegation", "BLOCKED", "SPYNEL_AGENT_ZERO_API_KEY absent")
    else:
        code, data = request(
            "/api/api_message",
            {"message": "hi", "agent_profile": "spynel", "async": True},
            api=True,
        )
        ok, final = (
            poll(str(data.get("context_id")), include_logs=True)
            if code == 200 and data.get("context_id")
            else (False, data)
        )
        diagnostic = json.dumps(final).lower()
        unexpected = any(
            marker in diagnostic
            for marker in ("memorize memories extension error", "rfc error", "chpasswd")
        )
        report(
            "chat", "PASS" if ok and final.get("result") and not unexpected else "FAIL"
        )
        report(
            "memory extension",
            "PASS" if not unexpected else "FAIL",
            "no unexpected memory/RFC/chpasswd diagnostics",
        )
        code, unknown = request(
            "/api/api_message",
            {
                "message": "inspect this",
                "agent_profile": "definitely-not-real",
                "async": True,
            },
            api=True,
        )
        report(
            "unknown profile safety",
            "PASS" if code == 404 and not unknown.get("context_id") else "FAIL",
        )
        code, delegated = request(
            "/api/plugins/_goal/delegate",
            {
                "objective": "Inspect runtime mode and return a read-only summary.",
                "agent_profile": "reviewer",
                "success_criteria": ["A read-only runtime summary is returned."],
                "idempotency_key": str(uuid.uuid4()),
                "idempotency_scope": "hf-smoke",
            },
            api=True,
        )
        if code == 200 and delegated.get("context_id") and delegated.get("goal_id"):
            ok, final = poll(str(delegated["context_id"]))
            if ok and final.get("result"):
                report("delegation", "PASS", "goal executed to completion")
            elif final.get("requires_attention") or final.get("status") in {
                "blocked",
                "interrupted",
                "partially_verified",
            }:
                report(
                    "delegation",
                    "BLOCKED",
                    "worker returned an explicit attention state",
                )
            else:
                report(
                    "delegation",
                    "FAIL",
                    "goal did not complete within the finite timeout",
                )
        else:
            report("delegation", "FAIL", "goal creation failed")
        discover()
    code, _ = request("/api/agent_profile_create", {})
    report("profile-create protection", "PASS" if code in {302, 401, 403} else "FAIL")
    if not API_KEY:
        for label in ("specialist profiles", "Spynel", "Paperclip skill"):
            report(label, "BLOCKED", "SPYNEL_AGENT_ZERO_API_KEY absent")
    mandatory = {
        "health",
        "settings",
        "runtime",
        "non-root",
        "root password capability",
        "API key",
        "compatible provider",
        "chat",
        "memory extension",
        "delegation",
        "unknown profile safety",
        "profile-create protection",
        "specialist profiles",
        "Spynel",
        "required skills",
        "spynel-dispatch",
        "paperclip-board-manager",
    }
    if target_sha:
        mandatory.add("deployed source SHA")
    if mode == "full":
        mandatory.add("settings persistence")
    return (
        1 if any(state != "PASS" for name, state in results if name in mandatory) else 0
    )


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--mode", choices=("basic", "authenticated", "full"), default="full"
    )
    parser.add_argument("--basic", action="store_true")
    parser.add_argument("--authenticated", action="store_true")
    parser.add_argument("--full", action="store_true")
    parser.add_argument("--expected-sha", default="", help="Expected GitHub source commit SHA")
    args = parser.parse_args()
    selected = (
        "basic" if args.basic else "authenticated" if args.authenticated else "full"
    )
    raise SystemExit(
        main(
            selected if any((args.basic, args.authenticated, args.full)) else args.mode,
            expected_sha=args.expected_sha,
        )
    )