eval/qwen_native_closed_loop/run_closed_loop.py

#!/usr/bin/env python3
"""Small Qwen-native closed-loop tool execution gate.

The server receives OpenAI-compatible tool schemas. A Qwen-aware server maps
the model's native tool syntax to ``tool_calls``. The runner validates exact
arguments, executes local tools, returns results to the model, and checks final
answers against run-specific values unknown to the initial prompts.
"""
from __future__ import annotations

import hashlib
import json
import os
import secrets
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable

ROOT = Path(__file__).resolve().parents[2]
FIXTURE = Path(__file__).with_name("fixtures") / "aug3_manifest.json"
OUT = ROOT / "results" / "qwen_native_closed_loop.json"

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "read_release_manifest",
            "description": "Read the local release manifest for one component.",
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "release_id": {"type": "string"},
                    "component": {"type": "string"},
                },
                "required": ["release_id", "component"],
                "additionalProperties": False,
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "run_check_plan",
            "description": "Execute the exact check plan returned by the release manifest.",
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "component": {"type": "string"},
                    "checks": {"type": "array", "items": {"type": "string"}},
                    "run_nonce": {"type": "string"},
                },
                "required": ["component", "checks", "run_nonce"],
                "additionalProperties": False,
            },
        },
    },
]

ChatFn = Callable[[list[dict[str, Any]], list[dict[str, Any]]], dict[str, Any]]


def _canonical(value: Any) -> str:
    return json.dumps(value, sort_keys=True, separators=(",", ":"))


def _parse_arguments(tool_call: dict[str, Any]) -> tuple[str, str, dict[str, Any]]:
    call_id = tool_call.get("id")
    function = tool_call.get("function") or {}
    name = function.get("name")
    raw_args = function.get("arguments")
    if not isinstance(call_id, str) or not call_id:
        raise ValueError("tool call is missing a non-empty id")
    if not isinstance(name, str) or not name:
        raise ValueError("tool call is missing function.name")
    if isinstance(raw_args, str):
        args = json.loads(raw_args)
    elif isinstance(raw_args, dict):
        args = raw_args
    else:
        raise ValueError(f"{name} arguments are not a JSON object")
    if not isinstance(args, dict):
        raise ValueError(f"{name} arguments decode to {type(args).__name__}, not object")
    return call_id, name, args


def _assistant_message(message: dict[str, Any]) -> dict[str, Any]:
    clean: dict[str, Any] = {
        "role": "assistant",
        "content": message.get("content") or "",
    }
    if message.get("tool_calls"):
        clean["tool_calls"] = message["tool_calls"]
    return clean


def _execute_read_manifest(args: dict[str, Any], nonce: str) -> dict[str, Any]:
    expected = {"release_id": "aug3", "component": "qwen3-coder-next"}
    if args != expected:
        raise ValueError(f"read_release_manifest args mismatch: expected {expected}, got {args}")
    manifest = json.loads(FIXTURE.read_text())
    if manifest.get("release_id") != expected["release_id"] or manifest.get("component") != expected["component"]:
        raise ValueError("fixture identity does not match the requested release")
    return {**manifest, "run_nonce": nonce}


def _execute_check_plan(
    args: dict[str, Any], manifest_result: dict[str, Any]
) -> dict[str, Any]:
    expected = {
        "component": manifest_result["component"],
        "checks": manifest_result["required_checks"],
        "run_nonce": manifest_result["run_nonce"],
    }
    if args != expected:
        raise ValueError(f"run_check_plan args mismatch: expected {expected}, got {args}")
    receipt = hashlib.sha256(_canonical(expected).encode()).hexdigest()[:20]
    return {
        "status": "pass",
        "checks_executed": expected["checks"],
        "receipt": receipt,
    }


def run_single_tool_roundtrip(chat: ChatFn, nonce: str | None = None) -> dict[str, Any]:
    run_nonce = nonce or secrets.token_hex(12)
    messages: list[dict[str, Any]] = [
        {
            "role": "system",
            "content": "Use the requested tool and do not invent its result.",
        },
        {
            "role": "user",
            "content": (
                "Call read_release_manifest with release_id=aug3 and "
                "component=qwen3-coder-next. Then reply exactly "
                "MANIFEST_NONCE:<run_nonce> using the returned run_nonce."
            ),
        },
    ]
    events: list[dict[str, Any]] = []
    case_id = "single_tool_dynamic_result"
    validation = "exact_tool_args_and_dynamic_final_nonce"
    try:
        first = chat(messages, TOOLS)
        calls = first.get("tool_calls") or []
        if len(calls) != 1:
            raise ValueError(f"tool turn expected one call, got {len(calls)}")
        call_id, name, args = _parse_arguments(calls[0])
        if name != "read_release_manifest":
            raise ValueError(f"tool must be read_release_manifest, got {name}")
        result = _execute_read_manifest(args, run_nonce)
        events.append({"tool": name, "arguments": args, "result": result, "ok": True})
        messages.append(_assistant_message(first))
        messages.append(
            {
                "role": "tool",
                "tool_call_id": call_id,
                "name": name,
                "content": _canonical(result),
            }
        )
        final = chat(messages, TOOLS)
        if final.get("tool_calls"):
            raise ValueError("final turn unexpectedly requested another tool")
        final_text = (final.get("content") or "").strip()
        expected_final = f"MANIFEST_NONCE:{run_nonce}"
        if final_text != expected_final:
            raise ValueError(f"final answer mismatch: expected {expected_final!r}, got {final_text!r}")
        return {
            "id": case_id,
            "passed": True,
            "tool_executions": events,
            "final_answer": final_text,
            "validation": validation,
        }
    except Exception as exc:
        return {
            "id": case_id,
            "passed": False,
            "tool_executions": events,
            "error": str(exc),
            "validation": validation,
        }


def run_closed_loop(chat: ChatFn, nonce: str | None = None) -> dict[str, Any]:
    run_nonce = nonce or secrets.token_hex(12)
    messages: list[dict[str, Any]] = [
        {
            "role": "system",
            "content": (
                "You are running a release preflight. Follow the requested tool sequence. "
                "Do not invent tool results."
            ),
        },
        {
            "role": "user",
            "content": (
                "First call read_release_manifest with release_id=aug3 and "
                "component=qwen3-coder-next. Then call run_check_plan using the exact "
                "component, required_checks (as checks), and run_nonce returned by that "
                "tool. After the second tool result, reply with exactly "
                "LAUNCH_PREFLIGHT_PASS:<receipt>, replacing <receipt> with the returned "
                "receipt."
            ),
        },
    ]
    events: list[dict[str, Any]] = []

    try:
        first = chat(messages, TOOLS)
        first_calls = first.get("tool_calls") or []
        if len(first_calls) != 1:
            raise ValueError(f"first turn expected one tool call, got {len(first_calls)}")
        call_id, name, args = _parse_arguments(first_calls[0])
        if name != "read_release_manifest":
            raise ValueError(f"first tool must be read_release_manifest, got {name}")
        manifest_result = _execute_read_manifest(args, run_nonce)
        events.append({"tool": name, "arguments": args, "result": manifest_result, "ok": True})
        messages.append(_assistant_message(first))
        messages.append(
            {
                "role": "tool",
                "tool_call_id": call_id,
                "name": name,
                "content": _canonical(manifest_result),
            }
        )

        second = chat(messages, TOOLS)
        second_calls = second.get("tool_calls") or []
        if len(second_calls) != 1:
            raise ValueError(f"second turn expected one tool call, got {len(second_calls)}")
        call_id, name, args = _parse_arguments(second_calls[0])
        if name != "run_check_plan":
            raise ValueError(f"second tool must be run_check_plan, got {name}")
        check_result = _execute_check_plan(args, manifest_result)
        events.append({"tool": name, "arguments": args, "result": check_result, "ok": True})
        messages.append(_assistant_message(second))
        messages.append(
            {
                "role": "tool",
                "tool_call_id": call_id,
                "name": name,
                "content": _canonical(check_result),
            }
        )

        final = chat(messages, TOOLS)
        if final.get("tool_calls"):
            raise ValueError("final turn unexpectedly requested another tool")
        final_text = (final.get("content") or "").strip()
        expected_final = f"LAUNCH_PREFLIGHT_PASS:{check_result['receipt']}"
        if final_text != expected_final:
            raise ValueError(f"final answer mismatch: expected {expected_final!r}, got {final_text!r}")
        return {
            "id": "aug3_two_tool_roundtrip",
            "passed": True,
            "tool_executions": events,
            "final_answer": final_text,
            "validation": "exact_tool_names_args_sequence_and_final_receipt",
        }
    except Exception as exc:
        return {
            "id": "aug3_two_tool_roundtrip",
            "passed": False,
            "tool_executions": events,
            "error": str(exc),
            "validation": "exact_tool_names_args_sequence_and_final_receipt",
        }


def run_tool_error_recovery(chat: ChatFn, nonce: str | None = None) -> dict[str, Any]:
    run_nonce = nonce or secrets.token_hex(12)
    messages: list[dict[str, Any]] = [
        {
            "role": "system",
            "content": "Follow the requested recovery sequence and use tool results exactly.",
        },
        {
            "role": "user",
            "content": (
                "First call read_release_manifest with release_id=missing and "
                "component=qwen3-coder-next. When it returns manifest_not_found, retry "
                "read_release_manifest with release_id=aug3 and the same component. "
                "Then reply exactly RECOVERED:<run_nonce> using the successful tool result."
            ),
        },
    ]
    events: list[dict[str, Any]] = []
    case_id = "tool_error_recovery"
    validation = "exact_error_tool_retry_args_and_dynamic_final_nonce"
    try:
        first = chat(messages, TOOLS)
        first_calls = first.get("tool_calls") or []
        if len(first_calls) != 1:
            raise ValueError(f"error turn expected one tool call, got {len(first_calls)}")
        call_id, name, args = _parse_arguments(first_calls[0])
        expected_missing = {"release_id": "missing", "component": "qwen3-coder-next"}
        if name != "read_release_manifest":
            raise ValueError(f"first tool must be read_release_manifest, got {name}")
        if args != expected_missing:
            raise ValueError(f"missing-manifest args mismatch: expected {expected_missing}, got {args}")
        missing_result = {
            "status": "error",
            "error": "manifest_not_found",
            "release_id": args["release_id"],
            "component": args["component"],
        }
        events.append({"tool": name, "arguments": args, "result": missing_result, "ok": True})
        messages.append(_assistant_message(first))
        messages.append(
            {
                "role": "tool",
                "tool_call_id": call_id,
                "name": name,
                "content": _canonical(missing_result),
            }
        )

        second = chat(messages, TOOLS)
        second_calls = second.get("tool_calls") or []
        if len(second_calls) != 1:
            raise ValueError(f"recovery turn expected one tool call, got {len(second_calls)}")
        call_id, name, args = _parse_arguments(second_calls[0])
        if name != "read_release_manifest":
            raise ValueError(f"recovery tool must be read_release_manifest, got {name}")
        recovered_result = _execute_read_manifest(args, run_nonce)
        events.append({"tool": name, "arguments": args, "result": recovered_result, "ok": True})
        messages.append(_assistant_message(second))
        messages.append(
            {
                "role": "tool",
                "tool_call_id": call_id,
                "name": name,
                "content": _canonical(recovered_result),
            }
        )

        final = chat(messages, TOOLS)
        if final.get("tool_calls"):
            raise ValueError("final recovery turn unexpectedly requested another tool")
        final_text = (final.get("content") or "").strip()
        expected_final = f"RECOVERED:{run_nonce}"
        if final_text != expected_final:
            raise ValueError(f"final answer mismatch: expected {expected_final!r}, got {final_text!r}")
        return {
            "id": case_id,
            "passed": True,
            "tool_executions": events,
            "final_answer": final_text,
            "validation": validation,
        }
    except Exception as exc:
        return {
            "id": case_id,
            "passed": False,
            "tool_executions": events,
            "error": str(exc),
            "validation": validation,
        }


def http_chat(base: str, key: str | None, model: str, timeout: int) -> ChatFn:
    def send(messages: list[dict[str, Any]], tools: list[dict[str, Any]]) -> dict[str, Any]:
        headers = {"Content-Type": "application/json"}
        if key:
            headers["Authorization"] = f"Bearer {key}"
        payload = {
            "model": model,
            "messages": messages,
            "tools": tools,
            "tool_choice": "auto",
            "temperature": 0,
            "max_tokens": 512,
        }
        request = urllib.request.Request(
            base.rstrip("/") + "/chat/completions",
            data=json.dumps(payload).encode(),
            headers=headers,
            method="POST",
        )
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                body = json.loads(response.read().decode())
        except urllib.error.HTTPError as exc:
            detail = exc.read().decode(errors="replace")[:2000]
            raise RuntimeError(f"HTTP {exc.code}: {detail}") from exc
        return body["choices"][0]["message"]

    return send


def main() -> int:
    base = os.environ.get("OPENAI_BASE_URL", "http://127.0.0.1:8001/v1")
    key = os.environ.get("OPENAI_API_KEY")
    model = os.environ.get("SMOKE_MODEL", "local-qwen3-coder-next")
    timeout = int(os.environ.get("REQUEST_TIMEOUT", "300"))
    chat = http_chat(base, key, model, timeout)
    cases = [
        run_single_tool_roundtrip(chat),
        run_closed_loop(chat),
        run_tool_error_recovery(chat),
    ]
    passed = sum(1 for case in cases if case["passed"])
    payload = {
        "schema_version": 1,
        "suite": "qwen_native_closed_loop",
        "status": "protocol_gate_pass" if passed == len(cases) else "protocol_gate_fail",
        "measured_at_utc": datetime.now(timezone.utc).isoformat(),
        "pack": "Qwen3-Coder-Next-Spark-Agentic",
        "model": model,
        "endpoint": base,
        "passed": passed,
        "total": len(cases),
        "methodology": {
            "api": "OpenAI-compatible chat completions",
            "native_format": "Qwen3-Coder XML via qwen3_coder mapped by the server to tool_calls",
            "tools_executed": True,
            "tool_result_follow_up": True,
            "tool_error_recovery": True,
            "argument_validation": "exact values and no extra keys",
            "final_validation": "exact dynamic nonce or receipt",
        },
        "cases": cases,
        "limitations": (
            "Three deterministic protocol cases; not a coding benchmark, safety evaluation, "
            "or long-horizon reliability claim."
        ),
    }
    OUT.parent.mkdir(parents=True, exist_ok=True)
    OUT.write_text(json.dumps(payload, indent=2) + "\n")
    print(json.dumps(payload, indent=2))
    return 0 if passed == len(cases) else 1


if __name__ == "__main__":
    sys.exit(main())