pvx/execution.py

"""Validate the packaged remote contract and model its surface decision.

Pure policy helpers for package/acceptance checks; no executor, HTTP client or agent runtime.
Cloud agents follow the equivalent execution-surface and mcp-workflow playbooks directly.
"""
from __future__ import annotations

import json
from pathlib import Path
from typing import Iterable
from urllib.parse import urlsplit

CONTRACT_PATH = "skills-shared/mcp-capabilities.json"
CREATE_OPERATIONS = {
    "image": {"account", "capabilities", "create_image", "generation_status"},
    "video": {"account", "capabilities", "create_video", "generation_status"},
    "voice": {"account", "audio_catalog", "create_voice", "audio_status"},
    "music": {"account", "audio_catalog", "create_music", "audio_status"},
    "template": {"account", "template", "create_template_media", "generation_status"},
}
VIDEO_OPERATIONS = {"video", "first_frame_video", "multi_reference_video", "reference",
                    "transition", "extend", "modify", "edit", "upscale", "motion-control", "last_frame_video"}
IMAGE_OPERATIONS = {"image", "image_reference"}
SUPPORTED_OPERATIONS = IMAGE_OPERATIONS | VIDEO_OPERATIONS | {"voice", "music", "template"}
RELEASE_TOOLS = set().union(*CREATE_OPERATIONS.values()) | {
    "import_asset", "import_url", "import_status", "render_media", "download_media",
    "audio_history", "task_status",
}
SUPPORTED_SKILLS = {
    "pixverse-studio", "pixverse-video-script", "pixverse-create-image", "pixverse-create-video",
    "pixverse-product-stills", "pixverse-character-sheet", "pixverse-brand-system",
    "pixverse-audio", "pixverse-voiceover", "pixverse-template",
}


def choose_surface(
    *, cli_state: str, skill: str, available_tools: Iterable[str], catalog: dict,
    operation: str = "image", planning_only: bool = False,
    in_flight: str | None = None, import_kind: str | None = None,
) -> dict[str, str]:
    """Model the documented gate using evidence supplied by the host, never guessed."""
    if cli_state not in {"usable", "recoverable", "unavailable", "unknown"}:
        raise ValueError("invalid CLI evidence state")
    if in_flight is not None:
        if in_flight not in {"cli", "mcp"}:
            raise ValueError("invalid in-flight transport")
        return {"surface": in_flight, "action": "recover", "reason": "preserve_submission"}
    if planning_only or skill == "pixverse-video-script":
        return {"surface": "chat", "action": "plan", "reason": "no_media_required"}
    if cli_state in {"usable", "recoverable"}:
        return {"surface": "cli", "action": "execute" if cli_state == "usable" else "repair", "reason": "cli_first"}
    if cli_state == "unknown":
        return {"surface": "undetermined", "action": "check", "reason": "need_structural_evidence"}
    if skill not in catalog["supported_skills"]:
        return {"surface": "mcp", "action": "stop", "reason": "unsupported_skill"}
    if skill == "pixverse-studio":
        return {"surface": "chat", "action": "discover", "reason": "supported_cloud_catalog"}
    if operation not in SUPPORTED_OPERATIONS:
        return {"surface": "mcp", "action": "stop", "reason": "unsupported_operation"}
    allowed = {
        "pixverse-create-video": VIDEO_OPERATIONS,
        "pixverse-audio": {"voice", "music"},
        "pixverse-voiceover": {"voice"},
        "pixverse-template": {"template"},
    }.get(skill, IMAGE_OPERATIONS)
    if operation not in allowed:
        return {"surface": "mcp", "action": "stop", "reason": "unsupported_skill_operation"}
    kind = "video" if operation in VIDEO_OPERATIONS else "image" if operation in IMAGE_OPERATIONS else operation
    required = set(CREATE_OPERATIONS[kind])
    if import_kind is not None:
        if import_kind not in {"attachment", "url"}:
            raise ValueError("invalid reference import kind")
        required |= {"import_status", "import_asset" if import_kind == "attachment" else "import_url"}
    if not required.issubset(set(available_tools)):
        return {"surface": "mcp", "action": "connect", "reason": "missing_callable_tools"}
    return {"surface": "mcp", "action": "execute", "reason": "supported_cloud_outcome"}


def membership_gate(member_type: object, *, fallback_consented: bool = False) -> str:
    """Mirror backend numeric levels; never reuse the CLI's label-only heuristics."""
    if isinstance(member_type, bool) or not isinstance(member_type, int) or member_type not in range(12):
        return "unknown"
    if member_type == 0:
        return "fallback" if fallback_consented else "upgrade_or_consent"
    return "server_entitlement_required"


def validate_mcp_contract(root: Path) -> list[str]:
    issues: list[str] = []
    try:
        config = json.loads((root / ".mcp.json").read_text())
        catalog = json.loads((root / CONTRACT_PATH).read_text())
        manifest = json.loads((root / ".codex-plugin/plugin.json").read_text())
        if manifest.get("mcpServers") != "./.mcp.json":
            issues.append("manifest must reference the single remote config")
        if set(config) != {"mcpServers"} or set(config["mcpServers"]) != {"pixverse"}:
            issues.append("exactly one PixVerse server is required")
        server = config["mcpServers"]["pixverse"]
        if set(server) != {"type", "url"} or server["type"] != "http":
            issues.append("remote config may contain only HTTP type and URL, never credentials")
        url = urlsplit(server["url"])
        if url.scheme != "https" or not url.hostname or url.path != "/mcp" or url.username or url.password or url.query or url.fragment:
            issues.append("endpoint must be an absolute HTTPS /mcp URL without credentials or API prefix")
        if server["url"] != "https://mcp.pixverse.ai/mcp":
            issues.append("release connection must use the production PixVerse endpoint")
        if catalog.get("schema_version") != 2 or catalog.get("app_only_tools") != []:
            issues.append("generation release requires schema v2 and model-callable downloads")
        media_ui = catalog.get("media_ui", {})
        if media_ui.get("optional") is not False or media_ui.get("delivery") != "required_live_host_media_card_with_url_fallback":
            issues.append("MCP Apps generation delivery must require live media cards with URL fallback")
        if media_ui.get("render_on") != "submission_ids_before_status_wait":
            issues.append("media cards must start immediately after submission, before status waits")
        if media_ui.get("card_scope") != "one_per_unique_image_video_audio_task_including_template_outputs":
            issues.append("media cards must cover every unique generation task including template outputs")
        if media_ui.get("fallback") != "renderer_unavailable_or_host_ui_failure":
            issues.append("URL-only delivery requires actual renderer or host UI unavailability")
        if media_ui.get("download") != "download_media_returns_temporary_url":
            issues.append("downloads must describe temporary URLs, not automatic local files")
        if catalog.get("optional_desktop_tools") != ["authorize_cli"]:
            issues.append("optional desktop login tool must remain separate from cloud production")
        if catalog.get("selection_policy") != "cli_first":
            issues.append("CLI must remain the preferred surface")
        supported = set(catalog["supported_skills"])
        others = set(catalog["other_skills"])
        public = {p.parent.name for p in (root / "skills").glob("*/SKILL.md")}
        if supported != SUPPORTED_SKILLS or supported & others or supported | others != public:
            issues.append("cloud catalog must classify every public skill exactly once with the reviewed generator entries")
        tools = set(catalog["tools"])
        if tools != RELEASE_TOOLS or len(catalog["tools"]) != len(tools):
            issues.append("tool inventory differs from the production generation release")
        required = catalog.get("required_tools", {})
        if set(required) != supported or any(not set(names) <= tools for names in required.values()):
            issues.append("every supported skill needs valid required-tool declarations")
        for skill in public:
            body = (root / "skills" / skill / "SKILL.md").read_text()
            gate = body.find("## Execution Surface")
            if gate < 0 or gate > body.find("\n# ") + 160:
                issues.append(f"{skill}: execution gate must precede workflow instructions")
            if "../../skills-shared/execution-surface.md" not in body:
                issues.append(f"{skill}: missing execution surface contract")
            if skill in others and "**Desktop required: do not activate this production skill in MCP mode.**" not in body:
                issues.append(f"{skill}: missing desktop-only activation gate")
            if skill in supported and f"use its **{skill}** recipe only" not in body:
                issues.append(f"{skill}: missing cloud recipe gate")
    except (OSError, KeyError, TypeError, ValueError, AttributeError) as exc:
        issues.append(f"invalid MCP contract: {exc}")
    return issues