Spaces:
Paused
Paused
| from contextlib import asynccontextmanager | |
| import datetime | |
| import os | |
| from pathlib import Path | |
| import subprocess | |
| import httpx | |
| from fastapi import FastAPI, Request, Response | |
| from fastapi.responses import HTMLResponse | |
| children: dict[str, subprocess.Popen] = {} | |
| def resolve_data_dir() -> Path: | |
| data_dir = Path(os.environ.get("HERMES_DATA_DIR", "/data")) | |
| if data_dir.is_dir() and os.access(data_dir, os.W_OK): | |
| return data_dir | |
| fallback = Path.home() / "hermes_data" | |
| fallback.mkdir(parents=True, exist_ok=True) | |
| os.environ["HERMES_DATA_DIR"] = str(fallback) | |
| return fallback | |
| def write_initial_config(data_dir: Path) -> None: | |
| env_path = data_dir / ".env" | |
| config_path = data_dir / "config.yaml" | |
| if not env_path.exists(): | |
| env_path.write_text( | |
| "OPENAI_API_KEY={}\nANTHROPIC_API_KEY={}\n".format( | |
| os.environ.get("OPENAI_API_KEY", ""), | |
| os.environ.get("ANTHROPIC_API_KEY", ""), | |
| ) | |
| ) | |
| if not config_path.exists(): | |
| config_path.write_text( | |
| "model: claude-opus-4-5\ndata_dir: {}\n".format(data_dir) | |
| ) | |
| def start_child(name: str, command: list[str], env: dict[str, str]) -> None: | |
| proc = children.get(name) | |
| if proc and proc.poll() is None: | |
| return | |
| log_dir = Path("/app/logs") | |
| log_dir.mkdir(parents=True, exist_ok=True) | |
| out = (log_dir / f"{name}.log").open("ab", buffering=0) | |
| err = (log_dir / f"{name}_err.log").open("ab", buffering=0) | |
| children[name] = subprocess.Popen(command, stdout=out, stderr=err, env=env) | |
| async def lifespan(app: FastAPI): | |
| data_dir = resolve_data_dir() | |
| write_initial_config(data_dir) | |
| env = os.environ.copy() | |
| env["HERMES_HOME"] = str(data_dir) | |
| env["HERMES_DATA_DIR"] = str(data_dir) | |
| env.setdefault("API_SERVER_ENABLED", "true") | |
| env.setdefault("API_SERVER_HOST", "127.0.0.1") | |
| env.setdefault("API_SERVER_PORT", "8642") | |
| start_child("hermes_gateway", ["hermes", "gateway", "run", "--replace"], env) | |
| yield | |
| for proc in children.values(): | |
| if proc.poll() is None: | |
| proc.terminate() | |
| app = FastAPI(lifespan=lifespan) | |
| async def root(): | |
| data_dir = Path(os.environ.get("HERMES_DATA_DIR", "/data")) | |
| storage_ok = data_dir.is_dir() and os.access(data_dir, os.W_OK) | |
| gateway_ok = children.get("hermes_gateway") and children["hermes_gateway"].poll() is None | |
| ts = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC") | |
| hermes_state = "Running" if gateway_ok else "Starting..." | |
| storage_state = f"{data_dir} writable" if storage_ok else "Local fallback" | |
| return f""" | |
| <!DOCTYPE html><html lang=en><head><meta charset=UTF-8> | |
| <meta name=viewport content="width=device-width,initial-scale=1"> | |
| <title>TestClaw</title> | |
| <style> | |
| :root{{--bg:#101112;--surface:#181b1d;--border:#30363a;--text:#e5e2dc;--muted:#9a958c;--accent:#5aa6a8;--green:#72b65a;--yellow:#e4b84e}} | |
| *{{box-sizing:border-box;margin:0;padding:0}} | |
| body{{background:var(--bg);color:var(--text);font-family:Arial,sans-serif;min-height:100vh;display:flex;flex-direction:column}} | |
| header,footer{{padding:1rem 1.5rem;border-color:var(--border);display:flex;justify-content:space-between;gap:1rem}} | |
| header{{border-bottom:1px solid var(--border)}}footer{{border-top:1px solid var(--border);color:var(--muted);font-size:.8rem}} | |
| .logo{{color:var(--accent);font-weight:700}} | |
| main{{padding:1.5rem;display:grid;grid-template-columns:repeat(auto-fit,minmax(250px,1fr));gap:1rem;flex:1}} | |
| .card{{background:var(--surface);border:1px solid var(--border);border-radius:8px;padding:1rem}} | |
| .label{{font-size:.75rem;text-transform:uppercase;color:var(--muted);margin-bottom:.75rem}} | |
| .status{{display:flex;align-items:center;gap:.5rem;font-weight:700}} | |
| .dot{{width:8px;height:8px;border-radius:50%}}.green{{background:var(--green)}}.yellow{{background:var(--yellow)}} | |
| .meta{{font-size:.85rem;color:var(--muted);line-height:1.5;margin-top:.5rem}} | |
| a{{color:var(--accent)}} | |
| </style></head><body> | |
| <header><span class=logo>TestClaw</span><span>Hermes autonomous research container</span></header> | |
| <main> | |
| <section class=card><div class=label>Hermes Gateway</div><div class=status><span class="dot {'green' if gateway_ok else 'yellow'}"></span>{hermes_state}</div><div class=meta>API proxy: <a href=/api/>/api/</a></div></section> | |
| <section class=card><div class=label>Control Surface</div><div class=status><span class="dot green"></span>Available</div><div class=meta>TestClaw status and routing are served directly by this Space.</div></section> | |
| <section class=card><div class=label>Storage</div><div class=status><span class="dot {'green' if storage_ok else 'yellow'}"></span>{storage_state}</div><div class=meta>Attach persistent storage at /data for durable memory.</div></section> | |
| </main> | |
| <footer><span>F555/testclaw</span><span>{ts}</span></footer> | |
| </body></html>""" | |
| async def health(): | |
| return { | |
| "status": "ok", | |
| "hermes_gateway": bool(children.get("hermes_gateway") and children["hermes_gateway"].poll() is None), | |
| "ts": datetime.datetime.utcnow().isoformat(), | |
| } | |
| async def proxy_request(request: Request, target_base: str, path: str) -> Response: | |
| url = f"{target_base}/{path}" | |
| if request.url.query: | |
| url = f"{url}?{request.url.query}" | |
| headers = { | |
| key: value | |
| for key, value in request.headers.items() | |
| if key.lower() not in {"host", "content-length"} | |
| } | |
| body = await request.body() | |
| async with httpx.AsyncClient(timeout=60.0, follow_redirects=False) as client: | |
| upstream = await client.request( | |
| request.method, | |
| url, | |
| headers=headers, | |
| content=body, | |
| ) | |
| response_headers = { | |
| key: value | |
| for key, value in upstream.headers.items() | |
| if key.lower() not in {"content-encoding", "transfer-encoding", "connection"} | |
| } | |
| return Response(upstream.content, status_code=upstream.status_code, headers=response_headers) | |
| async def api_proxy(request: Request, path: str = ""): | |
| return await proxy_request(request, "http://127.0.0.1:8642", path) | |
| async def hermes_proxy(request: Request, path: str = ""): | |
| return Response( | |
| "Hermes web dashboard is disabled in the slim Space image. Use /api/ for gateway access.", | |
| status_code=503, | |
| media_type="text/plain", | |
| ) | |