agents/openenv_server/__main__.py

from __future__ import annotations

import argparse
from pathlib import Path

import uvicorn
from openenv.core.env_server import create_fastapi_app

from agents.hero.env import HeroEnvironment
from agents.hero.schema import HeroObservation, HeroServerAction
from agents.master.env import DMEnvironment
from agents.master.sample import load_world
from agents.master.schema import DMAction, DMObservation
from agents.shared.runtime import build_interface_adapter, resolve_interface_config


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Serve dungeon environments over OpenEnv HTTP/WebSocket APIs.")
    parser.add_argument("role", choices=["dm", "hero"])
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int)
    parser.add_argument("--world", type=Path, help="Optional world definition JSON for hero serving.")
    parser.add_argument("--artifacts-root", type=Path)
    parser.add_argument("--max-concurrent-envs", type=int, default=1)
    parser.add_argument("--interface-provider", choices=["strict", "simple", "gemini"])
    parser.add_argument("--interface-model")
    parser.add_argument("--interface-narrate", action="store_true")
    parser.add_argument(
        "--translate-corporate-env",
        action="store_true",
        help="Rewrite hero-facing observations into a corporate app metaphor and map translated commands back through Gemini.",
    )
    args = parser.parse_args(argv)

    interface_config = resolve_interface_config(
        provider=args.interface_provider,
        model_name=args.interface_model,
        narrate_observations=args.interface_narrate,
        translation_mode="corporate_app" if args.translate_corporate_env else None,
    )

    if args.role == "dm":
        env_factory = lambda: DMEnvironment(
            artifacts_root=args.artifacts_root,
            interface_adapter=build_interface_adapter(interface_config),
        )
        action_cls = DMAction
        observation_cls = DMObservation
        default_port = 8001
    else:
        world_input = load_world(str(args.world)) if args.world is not None else None
        env_factory = lambda: HeroEnvironment(
            artifacts_root=args.artifacts_root,
            world_input=world_input,
            interface_adapter=build_interface_adapter(interface_config),
        )
        action_cls = HeroServerAction
        observation_cls = HeroObservation
        default_port = 8002

    app = create_fastapi_app(
        env=env_factory,
        action_cls=action_cls,
        observation_cls=observation_cls,
        max_concurrent_envs=args.max_concurrent_envs,
    )
    uvicorn.run(app, host=args.host, port=args.port or default_port)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())