Project-Infinity/start_jarvis.py at main · warheart1984-ctrl/Project-Infinity · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
from __future__ import annotations

import os
import time
import urllib.request

from aais.launcher import main


LEGACY_BACKEND_PORT = 5000
CANONICAL_BACKEND_PORT = 8000
LEGACY_BACKEND_URL = f"http://127.0.0.1:{LEGACY_BACKEND_PORT}"
LEGACY_BACKEND_HEALTH_URL = f"{LEGACY_BACKEND_URL}/health"
CANONICAL_BACKEND_URL = f"http://127.0.0.1:{CANONICAL_BACKEND_PORT}"
CANONICAL_BACKEND_HEALTH_URL = f"{CANONICAL_BACKEND_URL}/health"
BACKEND_TIMEOUT_SECONDS = 45


def build_frontend_env(backend_url: str, base_env: dict[str, str] | None = None) -> dict[str, str]:
    env = dict(base_env or os.environ.copy())
    env["VITE_API_URL"] = str(backend_url)
    env["REACT_APP_API_URL"] = str(backend_url)
    return env


def http_ready(url: str) -> bool:
    try:
        with urllib.request.urlopen(url, timeout=2) as response:
            return 200 <= response.status < 500
    except Exception:
        return False


def resolve_existing_backend(kind: str | None = None) -> dict[str, str] | None:
    targets = [
        {
            "backend_kind": "canonical",
            "backend_mode": "canonical",
            "backend_runtime": "already_running",
            "backend_url": CANONICAL_BACKEND_URL,
            "health_url": CANONICAL_BACKEND_HEALTH_URL,
        },
        {
            "backend_kind": "legacy",
            "backend_mode": "legacy",
            "backend_runtime": "already_running",
            "backend_url": LEGACY_BACKEND_URL,
            "health_url": LEGACY_BACKEND_HEALTH_URL,
        },
    ]

    for target in targets:
        if kind and target["backend_kind"] != kind:
            continue
        if http_ready(target["health_url"]):
            return {
                "status": "existing",
                "backend_kind": target["backend_kind"],
                "backend_mode": target["backend_mode"],
                "backend_runtime": target["backend_runtime"],
                "backend_url": target["backend_url"],
            }

    return None


def build_backend_candidates() -> list[dict[str, str]]:
    return [
        {
            "label": "AAIS canonical runtime",
            "kind": "canonical",
            "mode": "canonical",
            "backend_url": CANONICAL_BACKEND_URL,
            "health_url": CANONICAL_BACKEND_HEALTH_URL,
        }
    ]


def start_backend_candidate(
    candidate: dict[str, str],
    *,
    failures: list[str] | None = None,
) -> dict[str, str] | None:
    deadline = time.time() + BACKEND_TIMEOUT_SECONDS
    while time.time() < deadline:
        if http_ready(candidate["health_url"]):
            return {
                "status": "started",
                "backend_kind": candidate["kind"],
                "backend_mode": candidate["mode"],
                "backend_runtime": candidate["label"],
                "backend_url": candidate["backend_url"],
            }
        time.sleep(1)

    if failures is not None:
        failures.append(f"{candidate['label']} did not become healthy within {BACKEND_TIMEOUT_SECONDS}s.")
    return None


def ensure_backend() -> dict[str, str]:
    existing_canonical = resolve_existing_backend(kind="canonical")
    if existing_canonical:
        return existing_canonical

    existing_legacy = resolve_existing_backend(kind="legacy")
    candidates = build_backend_candidates()
    canonical_candidate = next(
        (candidate for candidate in candidates if candidate.get("kind") == "canonical"),
        None,
    )

    if existing_legacy and canonical_candidate:
        promoted_backend = start_backend_candidate(canonical_candidate)
        if promoted_backend:
            return promoted_backend
        return existing_legacy

    if existing_legacy:
        return existing_legacy

    failures: list[str] = []
    for candidate in candidates:
        backend_state = start_backend_candidate(candidate, failures=failures)
        if backend_state:
            return backend_state

    raise RuntimeError("Jarvis backend could not be started.\n\n" + "\n".join(failures))


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