scripts/demo_server.py

"""Local demo launcher: run the real server and mint a pairing code up front.

This is a development harness, not a second product. It builds the *actual*
:class:`ControlPlane`, the *actual* Gradio UI and the *actual* agent API. The
only difference from ``app.py`` is that it mints one pairing code before
launching and prints it, so a worker can be started from a script instead of by
clicking the button in the UI.

Useful when driving the app from automation (screenshots, smoke tests) or when
you just want a worker attached without going through the login flow.

    .venv\\Scripts\\python.exe scripts\\demo_server.py

Binding follows the same policy as ``app.py``: loopback unless
``DISTINCT_BIND_HOST`` says otherwise.
"""

from __future__ import annotations

import os
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from distinct_protocol.netpolicy import announce_bind, launch_kwargs  # noqa: E402
from distinct_server import ControlPlane  # noqa: E402
from distinct_server.identity import (  # noqa: E402
    NotConfigured,
    refuse_mocked_oauth,
)
from distinct_server.presentation import CSS, HEAD  # noqa: E402
from distinct_server.ui import build_app, distinct_theme  # noqa: E402


def main() -> int:
    # This serves the real application, so it refuses the same state
    # app.py refuses. It did not, and that was the gap: the guard the
    # project documents as load-bearing was absent from an entry point
    # that puts the real interface on a socket.
    try:
        refuse_mocked_oauth()
    except NotConfigured as reason:
        print(f"distinct: {reason}", file=sys.stderr)
        return 2
    control = ControlPlane()
    pairing = control.create_pairing_code("demo-operator")

    code_path = Path(__file__).resolve().parent.parent / "demo_pairing_code.txt"
    code_path.write_text(pairing.code, encoding="utf-8")

    print("=" * 72, flush=True)
    print(f"PAIRING CODE: {pairing.code}", flush=True)
    print(f"written to:   {code_path}", flush=True)
    print("=" * 72, flush=True)

    demo = build_app(control)
    network = launch_kwargs()
    announce_bind(str(network["server_name"]), int(network["server_port"]))
    demo.queue(default_concurrency_limit=16, max_size=256).launch(
        show_error=os.environ.get("DISTINCT_SHOW_ERRORS") == "1",
        theme=distinct_theme(),
        css=CSS,
        head=HEAD,
        **network,
    )
    return 0


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