Function _spawn_dashboard

dashboard/hermes-wrapper/plugin_api.py:209–292

Starts ``tracedecay dashboard`` and returns its base URL.

()

def _spawn_dashboard() -> str:
    """Starts ``tracedecay dashboard`` and returns its base URL."""
    binary = _find_tracedecay_bin()
    if not binary:
        raise HTTPException(
            status_code=503,
            detail=(
                "tracedecay binary not found. Install tracedecay or set "
                "TRACEDECAY_BIN / TRACEDECAY_DASHBOARD_URL."
            ),
        )
    project = _project_root()
    cmd = [
        binary,
        "dashboard",
        "--host",
        "127.0.0.1",
        "--port",
        "0",
        "--path",
        project,
    ]
    # Spawned on the dedicated long-lived thread so PDEATHSIG binds the
    # child's lifetime to the Hermes process, not a transient request thread.
    process = _spawn_pool.submit(
        subprocess.Popen,
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        env=_dashboard_env(),
        preexec_fn=_child_preexec if _libc is not None else None,  # noqa: PLW1509 — minimal prctl-only hook
    ).result(timeout=_SPAWN_TIMEOUT_SECONDS)

    # Single reader per pipe, for the child's whole lifetime: the stderr
    # drain keeps a bounded tail for error detail; the stdout reader parses
    # the URL line then KEEPS draining (a stopped reader would eventually
    # block the server on a full pipe buffer and 502 every proxied request).
    stderr_tail: deque = deque(maxlen=_STDERR_TAIL_LINES)
    threading.Thread(
        target=_drain_pipe, args=(process.stderr, stderr_tail), daemon=True
    ).start()

    # The first stdout line is stable: "tracedecay dashboard listening on <url>".
    # Extract the URL itself rather than depending on any surrounding text.
    url_ready = threading.Event()
    url_holder: dict[str, str] = {}

    def _read_stdout() -> None:
        if process.stdout is None:
            url_ready.set()
            return
        for line in process.stdout:
            stripped = line.strip()
            if not url_ready.is_set() and "listening on" in stripped:
                match = _LISTENING_URL_RE.search(stripped)
                if match:
                    url_holder["url"] = match.group(0).rstrip("/")