839 lines
32 KiB
Python
Executable File
839 lines
32 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Lifecycle manager for route-set-addressed Switchyard gateway processes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import time
|
|
from collections.abc import Mapping
|
|
from ipaddress import ip_address
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import urlsplit
|
|
|
|
from mmo_profiles import load_settings
|
|
from mmo_snapshot import load_snapshot
|
|
from mmo_state import (
|
|
ACTIVE_SESSION_STATUSES,
|
|
RECOVERABLE_JOB_STATUSES,
|
|
iter_job_records,
|
|
iter_session_records,
|
|
)
|
|
from mmo_util import (
|
|
atomic_write_json,
|
|
config_root,
|
|
file_lock,
|
|
filtered_environment,
|
|
http_json,
|
|
http_ready,
|
|
parse_env_file,
|
|
port_available,
|
|
process_matches,
|
|
process_start_token,
|
|
read_json,
|
|
state_root,
|
|
strict_json_loads,
|
|
terminate_process_group,
|
|
utc_now,
|
|
)
|
|
from mmo_version import MMO_SCHEMA_VERSION
|
|
|
|
_GATEWAY_PROCESSES: dict[int, subprocess.Popen[bytes]] = {}
|
|
_GATEWAY_ADMISSION_GRACE_SECONDS = 60.0
|
|
|
|
|
|
def _reap_gateway(pid: int | None) -> None:
|
|
"""Reap terminal gateway children without dropping live handles."""
|
|
|
|
if not pid:
|
|
return
|
|
process = _GATEWAY_PROCESSES.get(int(pid))
|
|
if process is None:
|
|
return
|
|
try:
|
|
process.wait(timeout=1.0)
|
|
except subprocess.TimeoutExpired:
|
|
return
|
|
except OSError:
|
|
pass
|
|
_GATEWAY_PROCESSES.pop(int(pid), None)
|
|
|
|
|
|
def gateways_root() -> Path:
|
|
root = state_root() / "gateways"
|
|
root.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
return root
|
|
|
|
|
|
def _validate_hash(value: str, label: str) -> str:
|
|
if len(value) != 64 or any(char not in "0123456789abcdef" for char in value):
|
|
raise ValueError(f"invalid {label}")
|
|
return value
|
|
|
|
|
|
def _gateway_key(snapshot_hash: str) -> str:
|
|
snapshot = load_snapshot(_validate_hash(snapshot_hash, "snapshot hash"))
|
|
return _validate_hash(
|
|
str(snapshot["manifest"].get("gateway_hash") or snapshot_hash),
|
|
"gateway hash",
|
|
)
|
|
|
|
|
|
def _gateway_dir_for_key(gateway_hash: str) -> Path:
|
|
path = gateways_root() / _validate_hash(gateway_hash, "gateway hash")
|
|
path.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
return path
|
|
|
|
|
|
def gateway_dir(snapshot_hash: str) -> Path:
|
|
"""Return the route-set gateway directory for a profile snapshot."""
|
|
|
|
return _gateway_dir_for_key(_gateway_key(snapshot_hash))
|
|
|
|
|
|
def _gateway_state_path_for_key(gateway_hash: str) -> Path:
|
|
return _gateway_dir_for_key(gateway_hash) / "gateway.json"
|
|
|
|
|
|
def gateway_state_path(snapshot_hash: str) -> Path:
|
|
return _gateway_state_path_for_key(_gateway_key(snapshot_hash))
|
|
|
|
|
|
def _binary(settings: Mapping[str, Any]) -> str:
|
|
configured = str(settings.get("switchyard_bin", "switchyard-server"))
|
|
resolved = shutil.which(configured)
|
|
if resolved:
|
|
return resolved
|
|
path = Path(configured).expanduser()
|
|
if path.is_file() and os.access(path, os.X_OK):
|
|
return str(path.resolve())
|
|
raise FileNotFoundError(f"Switchyard binary not found: {configured}")
|
|
|
|
|
|
def switchyard_version(binary: str | None = None) -> str:
|
|
"""Return the exact release reported by the configured gateway binary."""
|
|
|
|
resolved = binary or _binary(load_settings())
|
|
try:
|
|
result = subprocess.run(
|
|
[resolved, "--version"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
timeout=10,
|
|
env=filtered_environment(),
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
raise RuntimeError(f"unable to inspect Switchyard version at {resolved}: {exc}") from exc
|
|
output = (result.stdout + "\n" + result.stderr).strip()
|
|
match = re.fullmatch(r"switchyard-server ([0-9]+\.[0-9]+\.[0-9]+)", output)
|
|
if result.returncode != 0 or match is None:
|
|
raise RuntimeError(
|
|
"Switchyard did not report a supported semantic version: "
|
|
f"status={result.returncode}, output={output[-1000:]!r}"
|
|
)
|
|
return match.group(1)
|
|
|
|
|
|
def route_availability(snapshot: Mapping[str, Any]) -> dict[str, dict[str, Any]]:
|
|
"""Resolve current per-route availability without making workers startup-critical."""
|
|
|
|
credentials_path = config_root() / "credentials.env"
|
|
file_values = parse_env_file(credentials_path)
|
|
values = dict(file_values)
|
|
values.update({key: value for key, value in os.environ.items() if value})
|
|
result: dict[str, dict[str, Any]] = {}
|
|
for route_key, route in sorted(snapshot["resolved"]["routes"].items()):
|
|
credentials = list(route.get("credential_envs", []))
|
|
selected = next((name for name in credentials if values.get(name)), None)
|
|
available = selected is not None or not credentials
|
|
reason = None if available else "missing credential: " + "/".join(credentials)
|
|
if available and route.get("billing_mode") == "local" and route.get("base_url"):
|
|
parsed = urlsplit(str(route["base_url"]))
|
|
port = parsed.port or (443 if parsed.scheme == "https" else 80)
|
|
try:
|
|
with socket.create_connection((str(parsed.hostname), port), timeout=0.2):
|
|
pass
|
|
except OSError:
|
|
available = False
|
|
reason = f"local endpoint unavailable: {parsed.hostname}:{port}"
|
|
result[route_key] = {
|
|
"available": available,
|
|
"selected_credential_env": selected,
|
|
"reason": reason,
|
|
}
|
|
return result
|
|
|
|
|
|
def _credentials(snapshot: Mapping[str, Any]) -> tuple[dict[str, str], list[str]]:
|
|
credentials_path = config_root() / "credentials.env"
|
|
file_values = parse_env_file(credentials_path)
|
|
values = dict(file_values)
|
|
values.update({key: value for key, value in os.environ.items() if value})
|
|
root_agent = snapshot["resolved"]["agents"][snapshot["manifest"]["root_agent"]]
|
|
root_route = str(root_agent["route"])
|
|
groups = snapshot["manifest"].get("credential_groups", [])
|
|
|
|
resolved: dict[str, str] = {}
|
|
missing: list[str] = []
|
|
for group in groups:
|
|
alternatives = list(group.get("alternatives") or [group.get("target_env")])
|
|
source = next((name for name in alternatives if name and values.get(name)), None)
|
|
target = str(group.get("target_env") or alternatives[0])
|
|
if source is None:
|
|
route_key = str(group.get("route"))
|
|
if route_key == root_route:
|
|
missing.append("/".join(name for name in alternatives if name))
|
|
else:
|
|
# Switchyard resolves client environment keys when parsing its
|
|
# immutable route set. The supervisor prevents this sentinel
|
|
# client from receiving work through the availability overlay.
|
|
resolved[target] = f"mmo-unavailable-{route_key}"
|
|
continue
|
|
resolved[target] = values[source]
|
|
return resolved, missing
|
|
|
|
|
|
def _select_port(snapshot_hash: str, settings: Mapping[str, Any]) -> int:
|
|
minimum = int(settings.get("gateway_port_min", 42000))
|
|
maximum = int(settings.get("gateway_port_max", 51999))
|
|
if not 1024 <= minimum <= maximum <= 65535:
|
|
raise ValueError("invalid gateway port range in settings")
|
|
span = maximum - minimum + 1
|
|
start = int(snapshot_hash[:16], 16) % span
|
|
host = str(settings.get("gateway_host", "127.0.0.1"))
|
|
for offset in range(span):
|
|
port = minimum + ((start + offset) % span)
|
|
if port_available(host, port):
|
|
return port
|
|
raise RuntimeError("no free port is available in the configured gateway range")
|
|
|
|
|
|
def _health_url(host: str, port: int) -> str:
|
|
address = f"[{host}]" if ":" in host and not host.startswith("[") else host
|
|
return f"http://{address}:{port}/health"
|
|
|
|
|
|
def _base_url(host: str, port: int) -> str:
|
|
address = f"[{host}]" if ":" in host and not host.startswith("[") else host
|
|
return f"http://{address}:{port}/v1"
|
|
|
|
|
|
def _timestamp(value: Any, label: str) -> float:
|
|
if not isinstance(value, str) or not value:
|
|
raise ValueError(f"invalid {label}")
|
|
try:
|
|
parsed = dt.datetime.fromisoformat(value)
|
|
except ValueError as exc:
|
|
raise ValueError(f"invalid {label}") from exc
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=dt.UTC)
|
|
return parsed.timestamp()
|
|
|
|
|
|
def _read_gateway_state_by_key(gateway_hash: str) -> dict[str, Any] | None:
|
|
path = _gateway_state_path_for_key(gateway_hash)
|
|
if not path.is_file():
|
|
return None
|
|
try:
|
|
state = read_json(path)
|
|
except (OSError, ValueError) as exc:
|
|
return {
|
|
"gateway_hash": gateway_hash,
|
|
"status": "invalid",
|
|
"error": f"invalid gateway state: {type(exc).__name__}: {exc}",
|
|
}
|
|
if not isinstance(state, dict):
|
|
return {
|
|
"gateway_hash": gateway_hash,
|
|
"status": "invalid",
|
|
"error": "invalid gateway state: root must be an object",
|
|
}
|
|
schema_version = state.get("schema_version")
|
|
observed_switchyard_version = state.get("switchyard_version")
|
|
snapshot_hashes = state.get("snapshot_hashes", [])
|
|
profile_ids = state.get("profile_ids", [])
|
|
host = state.get("host")
|
|
port = state.get("port")
|
|
endpoint_identity_valid = False
|
|
if (
|
|
isinstance(host, str)
|
|
and isinstance(port, int)
|
|
and not isinstance(port, bool)
|
|
and 1 <= port <= 65535
|
|
):
|
|
with contextlib.suppress(ValueError):
|
|
endpoint_identity_valid = (
|
|
ip_address(host).is_loopback
|
|
and state.get("base_url") == _base_url(host, port)
|
|
and state.get("health_url") == _health_url(host, port)
|
|
)
|
|
if (
|
|
not isinstance(schema_version, int)
|
|
or isinstance(schema_version, bool)
|
|
or schema_version != MMO_SCHEMA_VERSION
|
|
or state.get("gateway_hash") != gateway_hash
|
|
or not isinstance(observed_switchyard_version, str)
|
|
or re.fullmatch(r"[0-9]+\.[0-9]+\.[0-9]+", observed_switchyard_version) is None
|
|
or not isinstance(state.get("pid"), int)
|
|
or isinstance(state.get("pid"), bool)
|
|
or not isinstance(state.get("process_start_token"), str)
|
|
or not isinstance(snapshot_hashes, list)
|
|
or not all(isinstance(value, str) for value in snapshot_hashes)
|
|
or not isinstance(profile_ids, list)
|
|
or not all(isinstance(value, str) for value in profile_ids)
|
|
or not endpoint_identity_valid
|
|
):
|
|
return {
|
|
"gateway_hash": gateway_hash,
|
|
"status": "invalid",
|
|
"error": "invalid gateway state identity or process metadata",
|
|
}
|
|
try:
|
|
_timestamp(state.get("last_used_at"), "gateway last-used timestamp")
|
|
except ValueError:
|
|
return {
|
|
"gateway_hash": gateway_hash,
|
|
"status": "invalid",
|
|
"error": "invalid gateway state last-used timestamp",
|
|
}
|
|
pid = state.get("pid")
|
|
if not process_matches(pid, state.get("process_start_token")):
|
|
state["status"] = "stopped"
|
|
return state
|
|
health = state.get("health_url")
|
|
state["status"] = "running" if isinstance(health, str) and http_ready(health) else "starting"
|
|
return state
|
|
|
|
|
|
def read_gateway_state(snapshot_hash: str) -> dict[str, Any] | None:
|
|
return _read_gateway_state_by_key(_gateway_key(snapshot_hash))
|
|
|
|
|
|
def _record_gateway_lease(
|
|
state: dict[str, Any], snapshot: Mapping[str, Any], gateway_hash: str
|
|
) -> dict[str, Any]:
|
|
snapshot_hash = str(snapshot["manifest"]["snapshot_hash"])
|
|
profile_id = str(snapshot["manifest"]["profile_id"])
|
|
state["gateway_hash"] = gateway_hash
|
|
state["snapshot_hashes"] = sorted(set(state.get("snapshot_hashes", [])) | {snapshot_hash})
|
|
state["profile_ids"] = sorted(set(state.get("profile_ids", [])) | {profile_id})
|
|
state["last_used_at"] = utc_now()
|
|
atomic_write_json(_gateway_state_path_for_key(gateway_hash), state)
|
|
return state
|
|
|
|
|
|
def ensure_gateway(snapshot_hash: str) -> dict[str, Any] | None:
|
|
snapshot = load_snapshot(snapshot_hash)
|
|
if not snapshot["manifest"].get("gateway_required"):
|
|
return None
|
|
gateway_hash = _validate_hash(
|
|
str(snapshot["manifest"].get("gateway_hash") or snapshot_hash),
|
|
"gateway hash",
|
|
)
|
|
settings = load_settings()
|
|
binary = _binary(settings)
|
|
observed_switchyard_version = switchyard_version(binary)
|
|
resolved_availability = route_availability(snapshot)
|
|
lock = gateways_root() / ".gateway.lock"
|
|
with file_lock(lock):
|
|
current = _read_gateway_state_by_key(gateway_hash)
|
|
if current and current.get("status") == "invalid":
|
|
raise RuntimeError(str(current.get("error")))
|
|
if (
|
|
current
|
|
and current.get("status") == "running"
|
|
and current.get("switchyard_version") == observed_switchyard_version
|
|
and current.get("route_availability") == resolved_availability
|
|
):
|
|
return _record_gateway_lease(current, snapshot, gateway_hash)
|
|
if current and process_matches(current.get("pid"), current.get("process_start_token")):
|
|
terminate_process_group(int(current["pid"]))
|
|
_reap_gateway(current.get("pid"))
|
|
credentials, missing = _credentials(snapshot)
|
|
if missing:
|
|
raise RuntimeError(
|
|
"missing root-route credentials for this profile: " + ", ".join(sorted(missing))
|
|
)
|
|
host = str(settings.get("gateway_host", "127.0.0.1"))
|
|
port = _select_port(gateway_hash, settings)
|
|
directory = _gateway_dir_for_key(gateway_hash)
|
|
log_path = directory / "gateway.log"
|
|
routing_log_path = directory / "routing.jsonl"
|
|
routes_path = Path(snapshot["directory"]) / "routes.toml"
|
|
command = [
|
|
binary,
|
|
"--config",
|
|
str(routes_path),
|
|
"--host",
|
|
host,
|
|
"--port",
|
|
str(port),
|
|
"--routing-log-file",
|
|
str(routing_log_path),
|
|
]
|
|
environment = filtered_environment(
|
|
allow_sensitive=credentials,
|
|
extra={
|
|
**credentials,
|
|
"RUST_LOG": os.environ.get("RUST_LOG", "switchyard_server=info,libsy=info"),
|
|
},
|
|
)
|
|
log_handle = log_path.open("ab", buffering=0)
|
|
try:
|
|
process = subprocess.Popen(
|
|
command,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=log_handle,
|
|
stderr=subprocess.STDOUT,
|
|
close_fds=True,
|
|
start_new_session=True,
|
|
env=environment,
|
|
cwd=directory,
|
|
)
|
|
finally:
|
|
log_handle.close()
|
|
_GATEWAY_PROCESSES[process.pid] = process
|
|
ready = False
|
|
try:
|
|
start_token = process_start_token(process.pid)
|
|
if start_token is None:
|
|
initial_returncode = process.poll()
|
|
tail = ""
|
|
with contextlib.suppress(OSError):
|
|
tail = log_path.read_text(encoding="utf-8", errors="replace")[-4000:]
|
|
error = (
|
|
f"Switchyard exited during startup with status {initial_returncode}"
|
|
if initial_returncode is not None
|
|
else "unable to fingerprint the Switchyard process"
|
|
)
|
|
raise RuntimeError(error + (f"\n{tail}" if tail else ""))
|
|
state = {
|
|
"schema_version": MMO_SCHEMA_VERSION,
|
|
"gateway_hash": gateway_hash,
|
|
"snapshot_hash": snapshot_hash,
|
|
"snapshot_hashes": [snapshot_hash],
|
|
"profile_id": snapshot["manifest"]["profile_id"],
|
|
"profile_ids": [snapshot["manifest"]["profile_id"]],
|
|
"switchyard_version": observed_switchyard_version,
|
|
"pid": process.pid,
|
|
"process_start_token": start_token,
|
|
"host": host,
|
|
"port": port,
|
|
"base_url": _base_url(host, port),
|
|
"health_url": _health_url(host, port),
|
|
"routes_path": str(routes_path),
|
|
"log_path": str(log_path),
|
|
"routing_log_path": str(routing_log_path),
|
|
"command": command,
|
|
"route_availability": resolved_availability,
|
|
"started_at": utc_now(),
|
|
"last_used_at": utc_now(),
|
|
"status": "starting",
|
|
}
|
|
atomic_write_json(_gateway_state_path_for_key(gateway_hash), state)
|
|
timeout = float(settings.get("gateway_start_timeout_seconds", 15))
|
|
deadline = time.monotonic() + max(1.0, timeout)
|
|
while time.monotonic() < deadline:
|
|
if process.poll() is not None:
|
|
break
|
|
if http_ready(state["health_url"], timeout=0.4):
|
|
state["status"] = "running"
|
|
state["ready_at"] = utc_now()
|
|
atomic_write_json(_gateway_state_path_for_key(gateway_hash), state)
|
|
ready = True
|
|
return state
|
|
time.sleep(0.1)
|
|
tail = ""
|
|
with contextlib.suppress(OSError):
|
|
tail = log_path.read_text(encoding="utf-8", errors="replace")[-4000:]
|
|
state["status"] = "failed"
|
|
state["finished_at"] = utc_now()
|
|
state["error"] = "Switchyard did not become healthy before the startup deadline"
|
|
atomic_write_json(_gateway_state_path_for_key(gateway_hash), state)
|
|
raise RuntimeError(state["error"] + (f"\n{tail}" if tail else ""))
|
|
finally:
|
|
if not ready:
|
|
terminate_process_group(process.pid)
|
|
_reap_gateway(process.pid)
|
|
|
|
|
|
def stop_gateway(snapshot_hash: str) -> dict[str, Any]:
|
|
gateway_hash = _gateway_key(snapshot_hash)
|
|
lock = gateways_root() / ".gateway.lock"
|
|
with file_lock(lock):
|
|
state = _read_gateway_state_by_key(gateway_hash)
|
|
if not state:
|
|
return {"snapshot_hash": snapshot_hash, "status": "not_found"}
|
|
if state.get("status") == "invalid":
|
|
raise RuntimeError(str(state.get("error")))
|
|
pid = state.get("pid")
|
|
if process_matches(pid, state.get("process_start_token")):
|
|
if not isinstance(pid, int) or isinstance(pid, bool):
|
|
raise RuntimeError("gateway state contains an invalid process id")
|
|
terminate_process_group(int(pid))
|
|
_reap_gateway(pid)
|
|
state["status"] = "stopped"
|
|
state["stopped_at"] = utc_now()
|
|
atomic_write_json(_gateway_state_path_for_key(gateway_hash), state)
|
|
return state
|
|
|
|
|
|
def gateway_status(snapshot_hash: str) -> dict[str, Any]:
|
|
state = read_gateway_state(snapshot_hash)
|
|
if state is None:
|
|
return {"snapshot_hash": snapshot_hash, "status": "not_started"}
|
|
return state
|
|
|
|
|
|
def list_gateways() -> list[dict[str, Any]]:
|
|
results: list[dict[str, Any]] = []
|
|
for directory in sorted(gateways_root().iterdir()):
|
|
if (
|
|
not directory.is_dir()
|
|
or len(directory.name) != 64
|
|
or any(char not in "0123456789abcdef" for char in directory.name)
|
|
):
|
|
continue
|
|
state = _read_gateway_state_by_key(directory.name)
|
|
if state:
|
|
results.append(state)
|
|
return results
|
|
|
|
|
|
def gateway_models(snapshot_hash: str) -> Any:
|
|
state = ensure_gateway(snapshot_hash)
|
|
if state is None:
|
|
return {"object": "list", "data": []}
|
|
return http_json(state["base_url"].rstrip("/") + "/models")
|
|
|
|
|
|
def dry_run_gateway(snapshot_hash: str) -> subprocess.CompletedProcess[str] | None:
|
|
snapshot = load_snapshot(snapshot_hash)
|
|
if not snapshot["manifest"].get("gateway_required"):
|
|
return None
|
|
settings = load_settings()
|
|
credentials, missing = _credentials(snapshot)
|
|
if missing:
|
|
raise RuntimeError("missing credentials: " + ", ".join(missing))
|
|
binary = _binary(settings)
|
|
environment = filtered_environment(
|
|
allow_sensitive=credentials,
|
|
extra={**credentials, "RUST_LOG": "error"},
|
|
)
|
|
return subprocess.run(
|
|
[binary, "--config", str(Path(snapshot["directory"]) / "routes.toml"), "--dry-run"],
|
|
env=environment,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
timeout=max(1.0, float(settings.get("gateway_start_timeout_seconds", 15))),
|
|
)
|
|
|
|
|
|
def route_telemetry(
|
|
metadata: Mapping[str, Any],
|
|
session: Mapping[str, Any],
|
|
events_path: Path,
|
|
) -> dict[str, Any]:
|
|
"""Extract observed serving-route identity from Codex and Switchyard evidence."""
|
|
|
|
provider_slugs: list[str] = []
|
|
endpoint_tags: list[str] = []
|
|
routing_attempts: list[int] = []
|
|
retry_counts: list[int] = []
|
|
fallback_indices: list[int] = []
|
|
|
|
def add_unique(target: list[str], value: Any) -> None:
|
|
if isinstance(value, str) and value.strip() and value not in target:
|
|
target.append(value)
|
|
|
|
def visit(value: Any, *, scope: str | None = None) -> None:
|
|
if isinstance(value, dict):
|
|
selected = value.get("selected")
|
|
status = value.get("status")
|
|
successful_attempt_record = (
|
|
isinstance(status, int) and not isinstance(status, bool) and 200 <= status < 300
|
|
)
|
|
for key, child in value.items():
|
|
normalized = str(key).casefold()
|
|
if normalized in {"openrouter_metadata", "routing"}:
|
|
visit(child, scope=normalized)
|
|
continue
|
|
if normalized in {
|
|
"provider_slug",
|
|
"serving_provider_slug",
|
|
"actual_provider_slug",
|
|
}:
|
|
add_unique(provider_slugs, child)
|
|
elif normalized == "provider" and scope in {
|
|
"openrouter_metadata",
|
|
"routing",
|
|
}:
|
|
if (
|
|
selected is True
|
|
or successful_attempt_record
|
|
or (scope == "routing" and selected is None)
|
|
):
|
|
add_unique(provider_slugs, child)
|
|
elif normalized == "provider_name" and (
|
|
scope in {"openrouter_metadata", "routing"}
|
|
or "upstream_id" in value
|
|
or "total_cost" in value
|
|
):
|
|
add_unique(provider_slugs, child)
|
|
elif normalized in {"provider_tag", "endpoint_tag", "serving_endpoint_tag"}:
|
|
add_unique(endpoint_tags, child)
|
|
elif normalized == "tag" and isinstance(value.get("selected"), bool):
|
|
if value.get("selected"):
|
|
add_unique(endpoint_tags, child)
|
|
elif (
|
|
normalized == "attempt"
|
|
and isinstance(child, int)
|
|
and not isinstance(child, bool)
|
|
and scope in {"openrouter_metadata", "routing"}
|
|
):
|
|
routing_attempts.append(child)
|
|
elif (
|
|
normalized in {"retry_count", "retries"}
|
|
and isinstance(child, int)
|
|
and not isinstance(child, bool)
|
|
):
|
|
retry_counts.append(max(0, child))
|
|
elif (
|
|
normalized == "fallback_index"
|
|
and isinstance(child, int)
|
|
and not isinstance(child, bool)
|
|
):
|
|
fallback_indices.append(max(0, child))
|
|
visit(child, scope=scope)
|
|
elif isinstance(value, list):
|
|
for child in value:
|
|
visit(child, scope=scope)
|
|
|
|
sources: list[str] = []
|
|
if events_path.is_file():
|
|
for line in events_path.read_text(encoding="utf-8", errors="replace").splitlines():
|
|
with contextlib.suppress(json.JSONDecodeError, ValueError):
|
|
visit(strict_json_loads(line))
|
|
sources.append("codex_events")
|
|
routing_path = session.get("gateway_routing_log_path")
|
|
if isinstance(routing_path, str) and Path(routing_path).is_file():
|
|
needles = {
|
|
str(metadata.get("model_key") or ""),
|
|
str(metadata.get("model") or ""),
|
|
str(metadata.get("route") or ""),
|
|
}
|
|
for line in Path(routing_path).read_text(encoding="utf-8", errors="replace").splitlines():
|
|
if not any(needle and needle in line for needle in needles):
|
|
continue
|
|
with contextlib.suppress(json.JSONDecodeError, ValueError):
|
|
visit(strict_json_loads(line))
|
|
sources.append("switchyard_routing_log")
|
|
successful_attempt = max(routing_attempts) if routing_attempts else None
|
|
observed_fallback_index = (
|
|
max(fallback_indices)
|
|
if fallback_indices
|
|
else max((attempt - 1 for attempt in routing_attempts if attempt >= 1), default=None)
|
|
)
|
|
retries = max(retry_counts) if retry_counts else None
|
|
requested_policy = metadata.get("requested_route_policy")
|
|
return {
|
|
"requested_policy": requested_policy,
|
|
"actual_serving_provider_slugs": provider_slugs,
|
|
"actual_serving_endpoint_tags": endpoint_tags,
|
|
"successful_attempt": successful_attempt,
|
|
"fallback_index": observed_fallback_index,
|
|
"retries": retries,
|
|
"retry_telemetry_complete": retries is not None,
|
|
"sources": sources,
|
|
"complete": requested_policy is None or bool(provider_slugs or endpoint_tags),
|
|
}
|
|
|
|
|
|
def route_telemetry_warnings(
|
|
metadata: Mapping[str, Any], observation: Mapping[str, Any]
|
|
) -> list[str]:
|
|
"""Interpret provider-route evidence without leaking that policy into workers."""
|
|
|
|
if metadata.get("requested_route_policy") is not None and not observation.get("complete"):
|
|
return [
|
|
"OpenRouter serving-provider telemetry is incomplete; requested policy is recorded "
|
|
"but the actual endpoint was not present in captured events"
|
|
]
|
|
return []
|
|
|
|
|
|
def _latest_record_timestamp(
|
|
record: Mapping[str, Any], fields: tuple[str, ...], label: str
|
|
) -> float:
|
|
observed: list[float] = []
|
|
for field in fields:
|
|
value = record.get(field)
|
|
if value is not None:
|
|
observed.append(_timestamp(value, f"{label} {field}"))
|
|
if not observed:
|
|
raise ValueError(f"{label} has no lifecycle timestamp")
|
|
return max(observed)
|
|
|
|
|
|
def _session_gateway_hash(session: Mapping[str, Any], *, verify_snapshot: bool) -> str | None:
|
|
recorded = session.get("gateway_hash")
|
|
if recorded is None:
|
|
if session.get("gateway_base_url") is not None:
|
|
raise ValueError("session has a gateway endpoint without a gateway identity")
|
|
return None
|
|
gateway_hash = _validate_hash(str(recorded), "session gateway hash")
|
|
if verify_snapshot:
|
|
expected = _gateway_key(str(session["snapshot_hash"]))
|
|
if gateway_hash != expected:
|
|
raise ValueError("session gateway identity does not match its snapshot")
|
|
return gateway_hash
|
|
|
|
|
|
def _session_admission_timestamp(session: Mapping[str, Any]) -> float:
|
|
return _timestamp(
|
|
session.get("last_active_at") or session.get("run_created_at") or session.get("created_at"),
|
|
"session admission timestamp",
|
|
)
|
|
|
|
|
|
def _job_admission_timestamp(job: Mapping[str, Any]) -> float:
|
|
return _timestamp(
|
|
job.get("recovery_requested_at") or job.get("created_at"),
|
|
"worker admission timestamp",
|
|
)
|
|
|
|
|
|
def _session_retains_gateway(session: Mapping[str, Any], now: float) -> bool:
|
|
status = session.get("status")
|
|
if status not in ACTIVE_SESSION_STATUSES:
|
|
return False
|
|
if process_matches(session.get("root_pid"), session.get("root_start_token")) or process_matches(
|
|
session.get("root_app_server_pid"), session.get("root_app_server_start_token")
|
|
):
|
|
return True
|
|
if status == "starting":
|
|
admitted_at = _session_admission_timestamp(session)
|
|
return now <= admitted_at + _GATEWAY_ADMISSION_GRACE_SECONDS
|
|
return False
|
|
|
|
|
|
def _job_retains_gateway(job: Mapping[str, Any], now: float) -> bool:
|
|
status = job.get("status")
|
|
if status not in RECOVERABLE_JOB_STATUSES:
|
|
return False
|
|
if process_matches(job.get("runner_pid"), job.get("runner_start_token")) or process_matches(
|
|
job.get("app_server_pid"), job.get("app_server_start_token")
|
|
):
|
|
return True
|
|
if status in {"queued", "starting", "recovering"}:
|
|
admitted_at = _job_admission_timestamp(job)
|
|
return now <= admitted_at + _GATEWAY_ADMISSION_GRACE_SECONDS
|
|
return False
|
|
|
|
|
|
def stop_idle_gateways() -> list[str]:
|
|
"""Stop route-set gateways only after every durable execution host releases them."""
|
|
|
|
settings = load_settings()
|
|
idle_seconds = int(settings.get("gateway_idle_timeout_seconds", 3600))
|
|
now = time.time()
|
|
active_gateway_hashes: set[str] = set()
|
|
last_consumer_at: dict[str, float] = {}
|
|
session_gateway_hashes: dict[str, str | None] = {}
|
|
try:
|
|
sessions = iter_session_records(strict=True)
|
|
jobs = iter_job_records(strict=True)
|
|
for session in sessions:
|
|
active = session.get("status") in ACTIVE_SESSION_STATUSES
|
|
gateway_hash = _session_gateway_hash(session, verify_snapshot=active)
|
|
session_id = str(session["session_id"])
|
|
session_gateway_hashes[session_id] = gateway_hash
|
|
if gateway_hash is None:
|
|
continue
|
|
release_at = _latest_record_timestamp(
|
|
session,
|
|
(
|
|
"finished_at",
|
|
"paused_at",
|
|
"suspended_at",
|
|
"detached_at",
|
|
"last_active_at",
|
|
"run_created_at",
|
|
"created_at",
|
|
),
|
|
f"session {session_id}",
|
|
)
|
|
if session.get("status") == "starting":
|
|
release_at = max(
|
|
release_at,
|
|
_session_admission_timestamp(session) + _GATEWAY_ADMISSION_GRACE_SECONDS,
|
|
)
|
|
last_consumer_at[gateway_hash] = max(
|
|
last_consumer_at.get(gateway_hash, 0.0), release_at
|
|
)
|
|
if _session_retains_gateway(session, now):
|
|
active_gateway_hashes.add(gateway_hash)
|
|
|
|
for job in jobs:
|
|
session_id = str(job["session_id"])
|
|
if session_id not in session_gateway_hashes:
|
|
if job.get("status") in RECOVERABLE_JOB_STATUSES:
|
|
raise ValueError(f"active worker {job['job_id']} refers to a missing session")
|
|
continue
|
|
gateway_hash = session_gateway_hashes[session_id]
|
|
if gateway_hash is None:
|
|
continue
|
|
release_at = _latest_record_timestamp(
|
|
job,
|
|
(
|
|
"finished_at",
|
|
"paused_at",
|
|
"suspended_at",
|
|
"last_progress_at",
|
|
"created_at",
|
|
),
|
|
f"worker {job['job_id']}",
|
|
)
|
|
if job.get("status") in {"queued", "starting", "recovering"}:
|
|
release_at = max(
|
|
release_at,
|
|
_job_admission_timestamp(job) + _GATEWAY_ADMISSION_GRACE_SECONDS,
|
|
)
|
|
last_consumer_at[gateway_hash] = max(
|
|
last_consumer_at.get(gateway_hash, 0.0), release_at
|
|
)
|
|
if _job_retains_gateway(job, now):
|
|
active_gateway_hashes.add(gateway_hash)
|
|
except (KeyError, OSError, RuntimeError, ValueError) as exc:
|
|
raise RuntimeError(
|
|
"cannot determine gateway idleness from invalid durable execution state"
|
|
) from exc
|
|
|
|
stopped: list[str] = []
|
|
for state in list_gateways():
|
|
gateway_hash = state.get("gateway_hash") or state.get("snapshot_hash")
|
|
if not gateway_hash or gateway_hash in active_gateway_hashes:
|
|
continue
|
|
if state.get("status") == "invalid":
|
|
raise RuntimeError(
|
|
"cannot determine gateway idleness from invalid gateway state: "
|
|
+ str(state.get("error"))
|
|
)
|
|
last_used_at = _timestamp(state.get("last_used_at"), "gateway last-used timestamp")
|
|
idle_since = max(last_used_at, last_consumer_at.get(str(gateway_hash), 0.0))
|
|
if now - idle_since >= idle_seconds and state.get("status") in {"running", "starting"}:
|
|
representative = str(state.get("snapshot_hash") or "")
|
|
if representative:
|
|
stop_gateway(representative)
|
|
stopped.append(representative)
|
|
return stopped
|