Launcher¶

def ensure_runtime_paths_initialized(backend_root: Optional[Path] = None) -> RuntimePaths:
    """Ensure runtime paths are initialized, defaulting to development mode.

    Args:
        backend_root: Explicit backend root directory. When omitted, defaults to
            the repository layout (``backend/``).

    Returns:
        RuntimePaths: Resolved runtime directories.
    """
    global _RUNTIME_PATHS
    if _RUNTIME_PATHS is None:
        root = backend_root or Path(__file__).resolve().parents[2]
        packaged_data = root / "data"
        _RUNTIME_PATHS = _compute_runtime_paths(
            mode="dev", backend_root=root, packaged_data_dir=packaged_data
        )
    return _RUNTIME_PATHS

def get_runtime_paths() -> RuntimePaths:
    """Return the current runtime paths.

    Returns:
        RuntimePaths: Current runtime path configuration.

    Raises:
        RuntimeError: If runtime paths were not initialized yet.
    """
    if _RUNTIME_PATHS is None:
        raise RuntimeError("Runtime paths not initialized")
    return _RUNTIME_PATHS

def initialize_runtime_paths(
    mode: str, backend_root: Path, packaged_data_dir: Optional[Path] = None
) -> RuntimePaths:
    """Initialize runtime paths for the current process.

    Args:
        mode: Runtime mode, typically ``"dev"`` or ``"prod"``.
        backend_root: Path to the backend root directory (contains ``src``).
        packaged_data_dir: Optional path to the read-only data payload bundled
            with the application. Only required for bundled builds.

    Returns:
        RuntimePaths: Resolved directory information for the process.

    Raises:
        ValueError: If runtime paths are initialized more than once or the mode
            is unsupported.
    """
    global _RUNTIME_PATHS
    if _RUNTIME_PATHS is not None:
        raise ValueError("Runtime paths already initialized")

    runtime_paths = _compute_runtime_paths(
        mode=mode, backend_root=backend_root, packaged_data_dir=packaged_data_dir
    )
    _RUNTIME_PATHS = runtime_paths
    return runtime_paths

def get_db_state() -> str:
    """Current DB health as seen by the watchdog: ok | recovering | failed."""
    return db_state

def start_watchdog(app: FastAPI) -> None:
    """Start the watchdog: attach the disconnect hook + spawn the recovery loop.

    Call AFTER ``init_database`` has bound the live engine (the hook attaches to
    ``core.db_engine``) and the checkpointer is open on ``app.state``.
    Idempotent: a second call is a no-op while the loop is running.
    """
    global _app, _loop_task, _wake, _event_loop, db_state
    if _loop_task is not None and not _loop_task.done():
        return
    _app = app
    _event_loop = asyncio.get_running_loop()
    _wake = asyncio.Event()
    db_state = DB_OK
    _register_error_listener()
    _loop_task = asyncio.create_task(_recovery_loop())

async def stop_watchdog() -> None:
    """Stop the watchdog cleanly: cancel + await the loop, detach the hook."""
    global _loop_task, _app, _wake, _event_loop
    task = _loop_task
    _loop_task = None
    if task is not None:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
    _remove_error_listener()
    _app = None
    _wake = None
    _event_loop = None
    logger.info("DB watchdog stopped")

def emit_event(payload: dict) -> None:
    """Print a structured JSON lifecycle event (newline-delimited) to stdout.

    A ``ts`` field (UTC ISO-8601 ms, Z) is stamped on every event; the
    caller's payload is not mutated.
    """
    print(json.dumps({**payload, "ts": _utc_timestamp()}), flush=True)

def emit_phase(phase: str) -> None:
    """Emit a startup-progress phase the frontend can surface on the loader.

    Phases are informational only — the frontend still gates readiness on the
    ``ready`` event / a confirming health check, never on a phase.
    """
    emit_event({"event": "phase", "phase": phase})

def harden_pg_hba(text: str) -> str:
    """Rewrite ``pg_hba.conf`` so every ``host`` rule authenticates with SCRAM.

    Pure text -> text: ``trust`` on a ``host*`` line becomes ``scram-sha-256``;
    ``local`` (Unix socket) rules, comments, blank lines, spacing and every
    other method are left byte-for-byte as they were. Idempotent.
    """
    return _TRUSTED_HOST_RULE.sub(r"\g<rule>scram-sha-256\g<rest>", text)

def postmaster_log_tail(data_dir: Path | str, max_lines: int = POSTMASTER_LOG_TAIL_LINES) -> str:
    """The embedded postmaster's own last lines, or ``""`` when it wrote none.

    Reads a bounded window from the end of the file (the log is unbounded),
    drops the partial line that window starts on, and redacts any password
    literal PostgreSQL echoed from a failing statement.

    Decoded defensively: Postgres writes its messages in the server encoding
    and in the operating system's language, so a byte that is not UTF-8 must
    cost a character, never the whole diagnostic. Never raises -- an absent or
    unreadable file simply has nothing to say.
    """
    log_file = Path(data_dir) / POSTMASTER_LOG_NAME
    try:
        size = log_file.stat().st_size
        start = max(0, size - POSTMASTER_LOG_WINDOW_BYTES)
        with log_file.open("rb") as handle:
            if start:
                handle.seek(start)
            raw = handle.read(POSTMASTER_LOG_WINDOW_BYTES)
    except OSError:
        return ""
    text = raw.decode("utf-8", errors="replace")
    if start:
        # The window landed mid-line; that fragment belongs to a line whose
        # beginning was not read.
        _, _, text = text.partition("\n")
    lines = [line for line in text.splitlines() if line.strip()]
    tail = _redact_passwords("\n".join(lines[-max_lines:]))
    return tail[-POSTMASTER_LOG_TAIL_CHARS:] if len(tail) > POSTMASTER_LOG_TAIL_CHARS else tail

def start_postgres(data_dir: Path | str) -> PostgresHandle:
    """Boot (or join) the embedded cluster and make the `erudi` DB ready.

    Idempotent: safe to call on an already-initialized data dir or while the
    cluster is already running (pgserver refcounts users of the data dir).

    Every failure below -- the postmaster that never came up, an admin
    connection refused, a database or an extension that could not be created
    -- leaves with the postmaster's own last lines attached
    (:func:`_attach_postmaster_log`), because the reason is in the server's
    log and nowhere else.
    """
    data_dir = Path(data_dir)
    try:
        data_dir.mkdir(parents=True, exist_ok=True)
        _prune_stale_handle_pids(data_dir)
        _recover_corrupt_pgdata(data_dir)

        server = _get_server_with_recovery(data_dir)
        # After initdb (the secret file must not pre-exist in PGDATA) and before
        # any other connection: the URLs handed out below all carry the password.
        password = _ensure_cluster_password(data_dir)
        admin_uri = _uri_with_password(server.get_uri(), password)
        _enforce_password_auth(admin_uri, data_dir, password)

        # CREATE DATABASE has no IF NOT EXISTS → guard on pg_database.
        # DB_NAME is an internal constant, never user input.
        with psycopg.connect(admin_uri, autocommit=True) as conn:
            exists = conn.execute(
                "SELECT 1 FROM pg_database WHERE datname = %s", (DB_NAME,)
            ).fetchone()
            if not exists:
                conn.execute(f'CREATE DATABASE "{DB_NAME}"')

        psycopg_url = _uri_for_db(admin_uri, DB_NAME)

        # pgvector extensions are per-database → create inside `erudi`, not the
        # admin DB probed above.
        with psycopg.connect(psycopg_url, autocommit=True) as conn:
            conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
    except BaseException as exc:
        _attach_postmaster_log(exc, data_dir)
        raise

    sqlalchemy_url = psycopg_url.replace("postgresql://", "postgresql+psycopg://", 1)
    logger.info(f"Embedded PostgreSQL ready (data_dir={data_dir})")
    return PostgresHandle(
        server=server,
        data_dir=data_dir,
        psycopg_url=psycopg_url,
        sqlalchemy_url=sqlalchemy_url,
    )

def stop_postgres(handle: PostgresHandle) -> None:
    """Stop the embedded cluster explicitly (deterministic shutdown order)."""
    # Announced before the call: pgserver's cleanup() stops the postmaster
    # (pg_ctl stop, then a kill), and a shutdown that hangs or throws there
    # would otherwise leave the log ending on the last thing the app did.
    logger.info(f"Stopping the embedded PostgreSQL cluster (data_dir={handle.data_dir})")
    handle.server.cleanup()
    logger.info("Embedded PostgreSQL stopped")

def ensure_runtime_paths_initialized(backend_root: Optional[Path] = None) -> RuntimePaths:
    """Ensure runtime paths are initialized, defaulting to development mode.

    Args:
        backend_root: Explicit backend root directory. When omitted, defaults to
            the repository layout (``backend/``).

    Returns:
        RuntimePaths: Resolved runtime directories.
    """
    global _RUNTIME_PATHS
    if _RUNTIME_PATHS is None:
        root = backend_root or Path(__file__).resolve().parents[2]
        packaged_data = root / "data"
        _RUNTIME_PATHS = _compute_runtime_paths(
            mode="dev", backend_root=root, packaged_data_dir=packaged_data
        )
    return _RUNTIME_PATHS

def get_runtime_paths() -> RuntimePaths:
    """Return the current runtime paths.

    Returns:
        RuntimePaths: Current runtime path configuration.

    Raises:
        RuntimeError: If runtime paths were not initialized yet.
    """
    if _RUNTIME_PATHS is None:
        raise RuntimeError("Runtime paths not initialized")
    return _RUNTIME_PATHS

def initialize_runtime_paths(
    mode: str, backend_root: Path, packaged_data_dir: Optional[Path] = None
) -> RuntimePaths:
    """Initialize runtime paths for the current process.

    Args:
        mode: Runtime mode, typically ``"dev"`` or ``"prod"``.
        backend_root: Path to the backend root directory (contains ``src``).
        packaged_data_dir: Optional path to the read-only data payload bundled
            with the application. Only required for bundled builds.

    Returns:
        RuntimePaths: Resolved directory information for the process.

    Raises:
        ValueError: If runtime paths are initialized more than once or the mode
            is unsupported.
    """
    global _RUNTIME_PATHS
    if _RUNTIME_PATHS is not None:
        raise ValueError("Runtime paths already initialized")

    runtime_paths = _compute_runtime_paths(
        mode=mode, backend_root=backend_root, packaged_data_dir=packaged_data_dir
    )
    _RUNTIME_PATHS = runtime_paths
    return runtime_paths

def bind_children_to_this_process() -> bool:
    """Create a kill-on-close Job Object and assign this process to it.

    Idempotent-ish: safe to call more than once (a second call just creates
    and leaks a second job doing the same thing), but callers should only call
    it once, early at startup and before spawning Postgres/llama-server, so
    every child inherits membership from the start.

    Returns True if the job was created and assigned, False if anything about
    it failed (already logged) -- the caller never needs to branch on this,
    it exists for tests.
    """
    global _job_handle

    if platform.system() != "Windows":
        return False

    try:
        kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)

        job = kernel32.CreateJobObjectW(None, None)
        if not job:
            logger.warning(
                f"CreateJobObjectW failed (error={ctypes.get_last_error()}); "
                "Postgres/llama-server may survive a hard-killed parent (#341)"
            )
            return False

        info = _JOBOBJECT_EXTENDED_LIMIT_INFORMATION()
        info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE

        ok = kernel32.SetInformationJobObject(
            wintypes.HANDLE(job),
            JobObjectExtendedLimitInformation,
            ctypes.byref(info),
            ctypes.sizeof(info),
        )
        if not ok:
            logger.warning(
                f"SetInformationJobObject failed (error={ctypes.get_last_error()}); "
                "Postgres/llama-server may survive a hard-killed parent (#341)"
            )
            kernel32.CloseHandle(wintypes.HANDLE(job))
            return False

        # GetCurrentProcess() returns a pseudo-handle (-1) valid for this call
        # without needing OpenProcess/PROCESS_SET_QUOTA|PROCESS_TERMINATE rights.
        current_process = kernel32.GetCurrentProcess()
        ok = kernel32.AssignProcessToJobObject(
            wintypes.HANDLE(job), wintypes.HANDLE(current_process)
        )
        if not ok:
            logger.warning(
                f"AssignProcessToJobObject failed (error={ctypes.get_last_error()}); "
                "Postgres/llama-server may survive a hard-killed parent (#341)"
            )
            kernel32.CloseHandle(wintypes.HANDLE(job))
            return False

        _job_handle = job  # keep alive: see module docstring
        logger.info(
            "Windows job object bound: Postgres/llama-server now die with this "
            "process even on a hard kill (#341)"
        )
        return True
    except Exception as exc:
        logger.warning(f"Could not set up the Windows job object (#341): {exc}")
        return False