Mujoco_WASM/decision_server/tests/website_fixture.py

"""Explicit opt-in E2E fixture: real gateway, loopback fake HTTP upstream; never deployed."""

import asyncio
import os
import tempfile
import time
from dataclasses import replace
from pathlib import Path

from aiohttp import web

from decision_server.providers import http, jev, openai
from decision_server.tests.test_service import plan_value
from decision_server.web_server import CATALOG, create_website_app


async def main():
    if os.environ.get("CADWORLD_E2E") != "1":
        raise RuntimeError("fixture_requires_explicit_opt_in")
    upstream = web.Application()

    async def respond(request):
        body = await request.json()
        import json

        if request.path == "/decisions":
            choices = {name: next(iter(q["criteria"])) for name, q in body["questions"].items()}
            if "grasp" in choices:
                evidence = json.loads(body["state"])["observation"]["evidence"]
                choices.update(
                    grasp="secure"
                    if evidence["secure"]
                    else "uncertain"
                    if any(v > 0.2 for v in evidence["fingerForces"])
                    else "empty",
                    diagnosis="none",
                    recovery="continue",
                )
                if "noul" in choices:
                    choices.update(noul="allow", score="good", reason="verified_progress")
            return web.json_response(
                {
                    "answers": {name: {"choice": choice} for name, choice in choices.items()},
                    "usage": {"input_tokens": 1},
                }
            )

        if '"action"' in json.dumps(body.get("text", {})):
            instruction = json.loads(body["input"])["instruction"]
            choices = {
                "前进0.5米": ("move", 0.5),
                "前进0.2米": ("move", 0.2),
                "后退0.2米": ("move", -0.2),
                "左转90度": ("turn", 90),
                "右转90度": ("turn", -90),
                "把方块搬到支撑台": ("pick_place", 0),
                "把方块搬到 B 区": ("pick_place", 0),
                "把方块搬到 (0.25,0.50) 米": ("pick_place", 0),
            }
            action, number = choices.get(instruction, ("clarify", 0))
            context = json.loads(body["input"])
            extra = {}
            if "sceneContext" in context:
                extra = dict(
                    version="lekiwi-language-v2",
                    objectId="none",
                    targetId="none",
                    position=[],
                    supportId="none",
                )
                if action == "pick_place":
                    extra.update(
                        objectId="block",
                        targetId="B" if "B" in instruction else "coordinates",
                        position=[] if "B" in instruction else [0.25, 0.50],
                        supportId="table",
                    )
            value = json.dumps(
                {
                    **extra,
                    "action": action,
                    "value": number,
                    "summary": instruction
                    if action != "clarify"
                    else "请说明左转还是右转，一次一个动作。",
                }
            )
        else:
            value = '{"ok":true}' if '"ok"' in str(body) else json.dumps(plan_value())
            context = json.loads(body.get("input", "{}"))
            if context.get("observation", {}).get("version") == "lekiwi-language-v2":
                from decision_server.protocol import LANGUAGE_PRECONDITIONS

                value = json.dumps(
                    dict(
                        version="lekiwi-language-v2",
                        objectId="block",
                        goalId="placement",
                        summary="测试规划",
                        steps=[
                            dict(skill=s, precondition=LANGUAGE_PRECONDITIONS[s], onFailure="stop")
                            for s in context["remaining"]
                        ],
                    )
                )
        if request.path == "/chat/completions":
            return web.json_response(
                {"choices": [{"finish_reason": "stop", "message": {"content": value}}]}
            )
        return web.json_response(
            {
                "status": "completed",
                "output": [
                    {"type": "message", "content": [{"type": "output_text", "text": value}]}
                ],
                "usage": {"input_tokens": 1},
            }
        )

    upstream.router.add_post("/{path:.*}", respond)
    runner = web.AppRunner(upstream, access_log=None)
    await runner.setup()
    site = web.TCPSite(runner, "127.0.0.1", 0)
    await site.start()
    port = site._server.sockets[0].getsockname()[1]

    async def fake_post(session, connection, path, payload):
        url = f"http://127.0.0.1:{port}" + ("/decisions" if not path else "")
        return await http.post(session, replace(connection, base_url=url), path, payload)

    openai.post = jev.post = fake_post
    # test_connection also references the bounded helper directly.
    from decision_server import server

    server.post = fake_post
    from decision_server.connections import Connection
    from decision_server.hosted_budget import HostedBudget

    quota_dir = tempfile.TemporaryDirectory()
    app = create_website_app(
        os.environ.get("CADWORLD_E2E_ORIGIN", "http://127.0.0.1:4180"),
        development=True,
        defaults={
            "llm": Connection(
                "responses", "https://api.deepseek.com", "deepseek-flash", "fixture-llm"
            ),
            "jev": Connection(
                "openrouter-decisions",
                "https://openrouter.ai/api/alpha/decisions",
                "typesafe/jev-1.13",
                "fixture-jev",
            ),
        },
        budget=HostedBudget(Path(quota_dir.name) / "budget.sqlite"),
    )
    catalog = app[CATALOG]
    catalog.models = {"fixture/structured": "Fixture structured model (not real)"}
    catalog.available = True

    async def refresh(_):
        catalog.checked_at = time.monotonic()

    catalog.refresh = refresh
    gateway = web.AppRunner(app, access_log=None, handler_cancellation=True)
    await gateway.setup()
    await web.TCPSite(
        gateway, "127.0.0.1", int(os.environ.get("CADWORLD_E2E_PORT", "8769"))
    ).start()
    try:
        await asyncio.Event().wait()
    finally:
        await gateway.cleanup()
        await runner.cleanup()
        quota_dir.cleanup()


if __name__ == "__main__":
    asyncio.run(main())