Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """Fail-closed preflight for the immutable Hermes upstream integration. | |
| The overlay still uses a small number of source-shape patches because upstream | |
| has no supported extension hook for these integration points. This verifier is | |
| run during the image build *before* dependency installation. A changed upstream | |
| shape therefore fails the build explicitly instead of producing a partially | |
| mounted application at runtime. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import sys | |
| from pathlib import Path | |
| class IntegrationShapeError(RuntimeError): | |
| pass | |
| def _read(path: Path) -> str: | |
| if not path.is_file(): | |
| raise IntegrationShapeError(f"required upstream file missing: {path}") | |
| return path.read_text(encoding="utf-8") | |
| def _require_one_of(text: str, variants: tuple[str, ...], label: str) -> None: | |
| if not any(variant in text for variant in variants): | |
| raise IntegrationShapeError(f"upstream integration marker missing: {label}") | |
| def verify(root: Path) -> list[str]: | |
| root = root.resolve() | |
| checks: list[str] = [] | |
| for name in ("package-lock.json", "uv.lock"): | |
| path = root / name | |
| if not path.is_file(): | |
| raise IntegrationShapeError(f"required deterministic lockfile missing: {path}") | |
| checks.append(name) | |
| web_server = _read(root / "hermes_cli" / "web_server.py") | |
| _require_one_of( | |
| web_server, | |
| ( | |
| 'app = FastAPI(title="Hermes Agent", version=__version__, lifespan=_lifespan)', | |
| "from tools.futures_dashboard_api import router as _futures_dashboard_router", | |
| ), | |
| "FastAPI application mount", | |
| ) | |
| _require_one_of( | |
| web_server, | |
| ( | |
| r'allow_origin_regex=r"^https?://(localhost|127\.0\.0\.1)(:\d+)?$"', | |
| r"https://[A-Za-z0-9-]+\.hf\.space", | |
| ), | |
| "CORS source shape", | |
| ) | |
| _require_one_of( | |
| web_server, | |
| ('@app.get("/api/status")', '@app.get("/health")'), | |
| "health/status route", | |
| ) | |
| checks.append("hermes_cli/web_server.py") | |
| toolsets = _read(root / "toolsets.py") | |
| _require_one_of(toolsets, ("TOOLSETS = {\n", '"futures_trading"'), "toolsets registry") | |
| checks.append("toolsets.py") | |
| public_paths = _read(root / "hermes_cli" / "dashboard_auth" / "public_paths.py") | |
| _require_one_of( | |
| public_paths, | |
| (' "/api/cron/fire",\n', '"/api/telegram/webhook"'), | |
| "dashboard public paths", | |
| ) | |
| checks.append("hermes_cli/dashboard_auth/public_paths.py") | |
| return checks | |
| def main(argv: list[str] | None = None) -> int: | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("root", nargs="?", default="/opt/hermes") | |
| args = parser.parse_args(argv) | |
| try: | |
| checks = verify(Path(args.root)) | |
| except (OSError, UnicodeError, IntegrationShapeError) as exc: | |
| print(f"ERROR: {exc}", file=sys.stderr) | |
| return 1 | |
| print(f"upstream_integration_preflight=PASS checks={len(checks)}") | |
| return 0 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |