Function _wait_until_ready

dashboard/hermes-wrapper/plugin_api.py:312–351

Blocks until the spawned engine answers /api/capabilities. Bounded by ``_READY_TIMEOUT_SECONDS``; on timeout (or child death) the child is reaped and a 503 with Retry-After is raised so clients know the engine is still warming up rather than broken.

(
    process: subprocess.Popen, url: str, project: str, stderr_tail: deque
)

def _wait_until_ready(
    process: subprocess.Popen, url: str, project: str, stderr_tail: deque
) -> None:
    """Blocks until the spawned engine answers /api/capabilities.

    Bounded by ``_READY_TIMEOUT_SECONDS``; on timeout (or child death) the
    child is reaped and a 503 with Retry-After is raised so clients know the
    engine is still warming up rather than broken.
    """
    deadline = time.monotonic() + _READY_TIMEOUT_SECONDS
    last_error = "no response"
    while time.monotonic() < deadline:
        if process.poll() is not None:
            error_lines = list(stderr_tail)[-5:]
            raise HTTPException(
                status_code=503,
                detail=(
                    "tracedecay dashboard exited during startup for project "
                    f"{project!r}: " + (" / ".join(error_lines) or "no output")
                ),
                headers={"Retry-After": "5"},
            )
        try:
            request = urllib.request.Request(f"{url}/api/capabilities", method="GET")
            with urllib.request.urlopen(request, timeout=2.0) as response:
                if response.status < 500:
                    return
                last_error = f"HTTP {response.status}"
        except Exception as exc:
            last_error = str(exc)
        time.sleep(_READY_POLL_INTERVAL_SECONDS)
    _terminate_process(process)
    raise HTTPException(
        status_code=503,
        detail=(
            f"tracedecay dashboard for project {project!r} did not become "
            f"ready within {_READY_TIMEOUT_SECONDS:.0f}s: {last_error}"
        ),
        headers={"Retry-After": "5"},
    )


def _shutdown() -> None: