"""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())