Spaces:
Running
Running
| """Authenticated maintenance residency routes. | |
| Maintenance can contain repository names, deployment state and console excerpts, so | |
| it is deliberately not exposed on the public /ui read surface. Mesh-aware clients | |
| use their existing owner/paid key through a server-side boundary. | |
| """ | |
| from __future__ import annotations | |
| from typing import Callable, Optional | |
| from .maintenance import maintenance | |
| from .maintenance_bootstrap import bootstrap | |
| from .maintenance_hf import start_hf_monitor | |
| from .maintenance_ingress import report_ci, verify_hf_owner_token | |
| from .xstrapp_operational import ensure_parent | |
| def build_maintenance_router(check_key: Callable, registry=None): | |
| from fastapi import APIRouter, Body, Header, HTTPException, Query | |
| router = APIRouter(prefix="/api/maintenance", tags=["Maintenance"]) | |
| ledger = maintenance(registry=registry) | |
| # Import the deploy-generated full git history before any live monitor appends | |
| # runtime events. Import is event-idempotent, so every restart is safe. | |
| router.mesh_maintenance_bootstrap = bootstrap(ledger) | |
| # Establish a known-good operational parent before the HF monitor can observe | |
| # an error. The first failure must split *from* something preserved, not make | |
| # the failing snapshot the system's accidental baseline. | |
| router.mesh_xstrapp_operational_parent = ensure_parent(ledger) | |
| monitor = start_hf_monitor(ledger) | |
| router.mesh_maintenance_ledger = ledger | |
| router.mesh_hf_monitor = monitor | |
| def authorize(x_signalmesh_key: Optional[str] = Header(default=None)): | |
| try: | |
| check_key(x_signalmesh_key, require_paid=True) | |
| except TypeError: | |
| check_key(x_signalmesh_key) | |
| except HTTPException: | |
| raise | |
| except Exception as exc: | |
| raise HTTPException(401, str(exc)) from exc | |
| return x_signalmesh_key | |
| def hf_owner(authorization: Optional[str]) -> str: | |
| scheme, _, token = (authorization or "").partition(" ") | |
| if scheme.casefold() != "bearer" or not token.strip(): | |
| raise HTTPException(401, "HF owner bearer token required") | |
| try: | |
| return verify_hf_owner_token(token.strip()) | |
| except Exception as exc: | |
| raise HTTPException(401, f"HF owner validation failed: {exc}") from exc | |
| def residency(task: str = Query(default="development", max_length=80), _key=Header(default=None, alias="X-SignalMesh-Key")): | |
| authorize(_key) | |
| return ledger.residency(task) | |
| def state(_key=Header(default=None, alias="X-SignalMesh-Key")): | |
| authorize(_key) | |
| return ledger.state() | |
| def read_ledger( | |
| after: int = Query(default=0, ge=0), | |
| limit: int = Query(default=200, ge=1, le=2000), | |
| _key=Header(default=None, alias="X-SignalMesh-Key"), | |
| ): | |
| authorize(_key) | |
| events = ledger.read(after=after, limit=limit) | |
| return {"schema": "signalmesh.maintenance/v1", "after": after, "events": events, "count": len(events), "cursor": ledger.state().get("cursor", 0)} | |
| def incident( | |
| seq: int, | |
| before: int = Query(default=20, ge=0, le=200), | |
| after: int = Query(default=5, ge=0, le=50), | |
| _key=Header(default=None, alias="X-SignalMesh-Key"), | |
| ): | |
| authorize(_key) | |
| return ledger.incident(seq, before=before, after=after) | |
| def agents_fq(task: str = Query(default="development", max_length=80), _key=Header(default=None, alias="X-SignalMesh-Key")): | |
| authorize(_key) | |
| return ledger.fq(task) | |
| def add_event(body: dict = Body(...), _key=Header(default=None, alias="X-SignalMesh-Key")): | |
| authorize(_key) | |
| return ledger.record( | |
| str(body.get("kind") or "external.maintenance"), | |
| source=str(body.get("source") or "api"), | |
| summary=str(body.get("summary") or ""), | |
| data=body.get("data"), | |
| level=str(body.get("level") or "info"), | |
| event_id=str(body.get("event_id") or ""), | |
| refs=body.get("refs") if isinstance(body.get("refs"), dict) else {}, | |
| ) | |
| def set_cell(frequency: str, body: dict = Body(...), _key=Header(default=None, alias="X-SignalMesh-Key")): | |
| authorize(_key) | |
| if not frequency.startswith("maint."): | |
| frequency = "maint." + frequency | |
| return ledger.set_cell( | |
| frequency, body.get("value"), source=str(body.get("source") or "api"), | |
| summary=str(body.get("summary") or frequency), level=str(body.get("level") or "info"), | |
| event_id=str(body.get("event_id") or ""), | |
| ) | |
| # Narrow telemetry ingress for GitHub Actions. The presented HF token is | |
| # validated against the configured HF owner, used only for that check, and is | |
| # never put in a signal, state cell, log, or ledger event. | |
| def ci_ingress( | |
| body: dict = Body(...), | |
| authorization: Optional[str] = Header(default=None, alias="Authorization"), | |
| ): | |
| owner = hf_owner(authorization) | |
| try: | |
| return report_ci(ledger, body, owner=owner) | |
| except ValueError as exc: | |
| raise HTTPException(422, str(exc)) from exc | |
| def refresh_hf(_key=Header(default=None, alias="X-SignalMesh-Key")): | |
| authorize(_key) | |
| active = router.mesh_hf_monitor | |
| if active is None: | |
| from .maintenance_hf import HFSpaceMonitor | |
| active = HFSpaceMonitor(ledger) | |
| return active.refresh() | |
| return router | |