python_tasks.py

"""Invoke tasks for the Python side of SecantusDB (the `secantus` package).

Imported by the root ``tasks.py`` (``from python_tasks import *``) so the
Python-server dev workflow — sync / test / test-one / perf / lint / fmt / serve /
docs / docs-serve / clean, plus ``py-gate`` (the full pre-commit sequence) and
``py-ship`` (gate → commit → push) — lives in one place, mirroring how the
``rust-*`` tasks live in ``rust_tasks.py``. ``import *`` brings the Task objects
into the root namespace for invoke discovery, so ``invoke test`` / ``invoke lint``
etc. are unchanged.

Scope: the pure-Python server (`src/secantus/**`) and its docs/tests. The
driver-conformance gauges (`validate-*`), the release pipeline
(`release-*`), and the bench/chaos harnesses stay in `tasks.py` — they're
cross-cutting (they target either server) rather than Python-server dev tasks.
"""

from __future__ import annotations

import getpass
import os
import re
import shlex
import shutil
import sys
import time

from invoke.context import Context
from invoke.tasks import task

#: `pty=True` makes invoke allocate a pseudo-terminal, which Windows does not
#: have: every task using it failed there with "your platform doesn't support
#: the 'pty' module" (`invoke sync` included, 2026-09-19). Off on Windows only.
PTY = sys.platform != "win32"


# Ruff is a pure-Python static check that needs neither the compiled `secantus`
# extension nor a synced project env, so `lint` / `fmt` run it through `uvx`
# (an ephemeral, cached tool env) instead of `uv run` — the latter would try to
# sync/build the project (a multi-minute WiredTiger compile) and fails outright
# in a fresh git worktree where `vendor/wiredtiger` isn't populated. Pin the
# version so local output matches CI's `ruff format --check`; keep this in sync
# with the `ruff==` pin in pyproject.toml's [project.optional-dependencies] dev.
_RUFF = "ruff@0.15.20"

# `docs` builds WT-free via `uv run --no-project` (autodoc imports `secantus`
# from src/ with `wiredtiger` mocked — see docs/conf.py), so it needs no project
# build and works in a bare git worktree. This overlay provides the Sphinx
# toolchain plus secantus's pure-Python runtime deps that autodoc must import;
# keep it in sync with pyproject's [project] dependencies (minus the compiled
# wiredtiger) and the docs entries of the dev extra.
_DOCS_DEPS = (
    "--with sphinx --with myst-parser --with furo "
    "--with pymongo --with shapely --with s2sphere --with python-dateutil"
)

# The known local-only failing test: a feature worktree's Rust crates link the
# /tmp WiredTiger build while the project wheel links its own, so this
# cross-server restore isn't byte-compatible locally (green in CI). Deselected
# from py-gate by default; pass --deselect "" to re-include it.
_LOCAL_DESELECT = "tests/test_rust_pitr_cross_server.py::test_python_restores_rust_data_to_a_mark"


@task
def sync(c: Context) -> None:
    """Sync the venv so the FULL suite can actually collect and run.

    Two traps this exists to avoid, both of which produce a green run that
    proved less than it looked:

    1. ``--extra dev`` alone omits the ``rust`` extra, so ``_secantus_core`` is
       not installed -- and the ~1700 engine-parity tests then do not collect
       at all (they ``pytest.importorskip`` it, deliberately, so pure-Python
       environments still work). The suite still exits 0, roughly 1700 tests
       short, and only comparing the collected count catches it. Worse, a
       partial sync PRUNES: syncing a subset once removed the whole Pelican
       toolchain the website tasks need. ``--all-extras`` is the dev default.

    2. ``uv sync`` does NOT rebuild ``secantus-core`` when its Rust source
       changes -- it is a path dependency, and uv reuses the cached build. Pull
       a commit that touches ``crates/secantus-core*`` and the venv keeps the
       OLD compiled extension, so the parity suites compare current Python
       against stale Rust. Observed as 31 phantom failures in
       ``test_rust_diff_parity.py`` on a perfectly green ``main``, which reads
       exactly like a regression. ``--reinstall-package`` busts that cache.
    """
    c.run(
        "uv sync --all-extras --reinstall-package secantus-core",
        pty=PTY,
    )


@task
def test(c: Context, k: str = "", verbose: bool = False) -> None:
    cmd = "uv run python -m pytest"
    if verbose:
        cmd += " -v"
    if k:
        # shlex.quote — `f"{k!r}"` uses Python repr() which wraps in
        # single quotes but doesn't escape embedded single quotes,
        # leaving a shell-injection hole on a CLI-supplied filter.
        cmd += f" -k {shlex.quote(k)}"
    c.run(cmd, pty=PTY)


@task(name="test-one")
def test_one(c: Context, nodeid: str) -> None:
    # `-n0 -o addopts=` runs serially: `-p no:xdist` fails because the
    # project's addopts still injects `-n auto --dist=loadgroup`, which xdist
    # then can't parse once its plugin is disabled. Clearing addopts and forcing
    # zero workers is the reliable single-test form.
    c.run(
        "uv run --no-sync python -m pytest -n0 -o addopts= -p no:cacheprovider "
        f"{shlex.quote(nodeid)}",
        pty=PTY,
    )


@task(name="perf")
def perf_task(c: Context) -> None:
    """Run the performance regression suite (serially, no xdist).

    Benchmarks fight for CPU under parallel workers, amplifying noise to
    the point that the gate becomes flappy — so this task forces serial
    execution and explicitly opts in to the ``perf`` marker excluded
    from the default ``invoke test``. Median time per workload is
    asserted against a hard upper bound calibrated for ``:memory:``
    storage on a quiet 2024-era arm64 mac. Lower the bounds in
    ``tests/test_perf_regression.py`` when an optimisation moves the
    floor.
    """
    c.run(
        "uv run python -m pytest -p no:xdist "
        "-o addopts= -m perf "
        "--benchmark-columns=min,median,max -v "
        "tests/test_perf_regression.py",
        pty=PTY,
    )


@task
def lint(c: Context) -> None:
    c.run(f"uvx {_RUFF} check src tests", pty=PTY)
    c.run(f"uvx {_RUFF} format --check src tests", pty=PTY)


@task
def fmt(c: Context) -> None:
    c.run(f"uvx {_RUFF} format src tests", pty=PTY)
    c.run(f"uvx {_RUFF} check --fix src tests", pty=PTY)


@task
def serve(c: Context, host: str = "127.0.0.1", port: int = 27017) -> None:
    c.run(
        f"uv run python -m secantus --host {shlex.quote(host)} --port {int(port)}",
        pty=PTY,
    )


@task
def docs(c: Context, builder: str = "html", clean: bool = False) -> None:
    # Build WT-free with `--no-project`: autodoc imports `secantus` from src/
    # (conf.py adds it to sys.path) with the compiled `wiredtiger` mocked
    # (conf.py autodoc_mock_imports), so no scikit-build-core / WiredTiger
    # compile runs. This is what lets `invoke docs` work in a bare git worktree
    # that never ran a project build. `_DOCS_DEPS` overlays the doc toolchain
    # plus secantus's pure-Python runtime deps.
    if clean:
        c.run("rm -rf docs/_build", pty=PTY)
    qb = shlex.quote(builder)
    c.run(
        f"uv run --no-project {_DOCS_DEPS} "
        f"sphinx-build -W --keep-going -b {qb} docs docs/_build/{qb}",
        pty=PTY,
    )


@task(name="docs-rust")
def docs_rust(c: Context, builder: str = "html", clean: bool = False) -> None:
    """Build the Rust-server docs tree (docs-rust/) with warnings-as-errors.

    Pure-markdown Sphinx tree (no autodoc, no secantus import) — the version
    is regex-read from crates/secantusdb/Cargo.toml, so this works in any
    bare worktree with no build at all.
    """
    if clean:
        c.run("rm -rf docs-rust/_build", pty=PTY)
    qb = shlex.quote(builder)
    c.run(
        f"uv run --no-project {_DOCS_DEPS} "
        f"sphinx-build -W --keep-going -b {qb} docs-rust docs-rust/_build/{qb}",
        pty=PTY,
    )


@task(name="docs-serve")
def docs_serve(c: Context, port: int = 8000) -> None:
    docs(c)
    c.run(
        f"uv run --no-sync python -m http.server {port} --directory docs/_build/html",
        pty=PTY,
    )


@task
def clean(c: Context) -> None:
    c.run(
        "rm -rf build dist *.egg-info .pytest_cache .ruff_cache .coverage htmlcov docs/_build",
        pty=PTY,
    )
    # Sweep leaked gauge tempdirs. Aborted runs of ``invoke validate-*``
    # leave ``secantus-<driver>-gauge-XXXXXX`` directories under the
    # system tempdir (``/var/folders/.../T`` on macOS, ``/tmp`` on
    # Linux). Each holds a WiredTiger store — tens of MB at minimum.
    # Reap anything older than an hour so an active gauge isn't
    # interrupted.
    import glob
    import os
    import shutil
    import tempfile
    import time

    cutoff = time.time() - 3600.0
    base = tempfile.gettempdir()
    candidates = glob.glob(os.path.join(base, "secantus-*-gauge-*"))
    swept = 0
    for path in candidates:
        try:
            if os.stat(path).st_mtime < cutoff:
                shutil.rmtree(path, ignore_errors=True)
                swept += 1
        except FileNotFoundError:
            continue
    if swept:
        print(f"clean: swept {swept} stale gauge tempdir(s) older than 1h under {base}")

    reaped, freed = _sweep_stale_pytest_tmp(base)
    if reaped:
        print(
            f"clean: reaped {reaped} abandoned pytest tempdir(s) "
            f"({freed / 1024**3:.1f} GiB) under {base}"
        )

    reaped, freed = _sweep_stale_probe_tmp(base)
    if reaped:
        print(
            f"clean: reaped {reaped} abandoned probe store(s) "
            f"({freed / 1024**3:.1f} GiB) under {base}"
        )


#: How many numbered pytest dirs to keep.
#:
#: ONE, not pytest's own default of three, because of what a run costs here. The
#: docstring below used to say "~1.7 GiB a run"; measured on 2026-09-22 a full
#: 16-worker run leaves **~104 GiB** -- the suite has grown by orders of
#: magnitude since that number was written, and nothing re-derived it. Three
#: retained runs is then ~300 GiB of WiredTiger homes, which is how a 935 GiB
#: box reached 50 MB free with the janitor working exactly as designed.
#:
#: Keeping the newest run is what post-mortem of a failure actually needs; the
#: two behind it have never been the thing anyone looked at.
_PYTEST_TMP_KEEP = 1


def _sweep_stale_pytest_tmp(base: str, *, measure: bool = True) -> tuple[int, int]:
    """Delete abandoned ``pytest-of-<user>/pytest-NNNN`` trees; return
    ``(count, bytes_freed)``.

    ``measure=False`` skips sizing the trees and reports ``0`` bytes. Sizing
    walks every file to produce a number for ``invoke clean``'s summary line,
    which doubles the I/O on a large backlog -- 37 GiB in one run directory has
    been seen. The automatic sweep in ``tests/conftest.py`` runs on every
    pytest start and wants none of that, so it opts out.

    This suite pins ``tmp_path_retention_policy = "all"`` on purpose — deleting
    a passed test's ``tmp_path`` mid-session races WiredTiger's background
    threads into ``WT_PANIC`` (see ``tests/conftest.py``) — so every run leaves
    its per-test WiredTiger databases behind: **~104 GiB a run** on a
    16-worker box (measured 2026-09-22; an earlier "~1.7 GiB" here was many
    thousands of tests out of date). Reclaiming them is pytest's job: it keeps
    the newest few and ``rmtree``s the rest.

    That janitor stops working the moment a run dies abnormally. pytest writes
    the owning PID into a ``.lock`` beside each dir and removes it in an
    ``atexit`` hook, so a run killed by SIGKILL — or by this suite's own stall
    watchdog, which calls ``os._exit(70)`` — leaves the lock behind, and pytest
    then treats that dir as live for a full ``LOCK_TIMEOUT`` (3 days). The
    backlog compounds: the bigger it gets the longer the exit-time cleanup
    takes, so more runs are killed mid-cleanup, each leaving another stale
    lock. One dev box reached 241 dirs / 391 GiB that way, and every pytest
    invocation on it paid an unbounded exit-time ``rmtree``.

    A lock's PID makes liveness decidable now instead of in three days: if the
    process is gone, the lock is stale and the tree is garbage. Dirs whose
    owner is still running are left strictly alone, as are the newest
    ``_PYTEST_TMP_KEEP``.
    """
    root = os.path.join(base, f"pytest-of-{getpass.getuser()}")
    if not os.path.isdir(root):
        return (0, 0)
    numbered = [
        os.path.join(root, name)
        for name in os.listdir(root)
        if re.fullmatch(r"pytest-\d+", name) and os.path.isdir(os.path.join(root, name))
    ]
    # Never touch symlinks (``pytest-current``) or the root itself.
    numbered = [p for p in numbered if not os.path.islink(p)]
    numbered.sort(key=lambda p: os.stat(p).st_mtime, reverse=True)

    reaped = freed = 0
    for path in numbered[_PYTEST_TMP_KEEP:]:
        if _pytest_tmp_owner_alive(path):
            continue
        try:
            if measure:
                freed += _dir_size(path)
            shutil.rmtree(path, ignore_errors=True)
            reaped += 1
        except OSError:
            continue
    return (reaped, freed)


#: Prefix of a probe's throwaway WiredTiger home.
#:
#: Duplicated from ``tools/probes/_servers.PROBE_TMP_PREFIX`` rather than
#: imported: that module imports ``pymongo`` at module scope, and this is
#: housekeeping that must work in an environment with no probe dependencies
#: installed (CI's slim ``storage-engine`` env is exactly that). A one-word
#: constant is a better dependency than a package import here.
#: ``tests/test_clean_pytest_tmp.py`` pins the two to each other.
_PROBE_TMP_PREFIX = "secantus-probe-"


def _sweep_stale_probe_tmp(base: str) -> tuple[int, int]:
    """Delete abandoned probe WiredTiger homes; return ``(count, bytes)``.

    A differential probe stands up real servers, so it needs a real WT home,
    and every one of the ~17 probes under ``tools/probes/`` used to take a bare
    ``tempfile.mkdtemp()`` that nothing ever deleted. Those are invisible to
    every janitor here: pytest only manages ``pytest-of-<user>/``, and the
    gauge sweep above only matches ``secantus-*-gauge-*``. One session left 385
    of them -- ~50 GiB.

    ``tools/probes/_servers.probe_store`` now deletes its store at exit, but
    that is best-effort and CANNOT be the whole answer: on Windows an open file
    cannot be deleted at all, so a probe that dies before stopping its server
    (the normal outcome when a probe is how you are chasing a bug) leaves the
    home behind with WiredTiger still holding it. That is why the store carries
    its creating PID -- liveness is decidable here rather than guessed from an
    mtime, the same rule the pytest sweep above uses.
    """
    reaped = freed = 0
    for name in os.listdir(base):
        if not name.startswith(_PROBE_TMP_PREFIX):
            continue
        path = os.path.join(base, name)
        if not os.path.isdir(path) or os.path.islink(path):
            continue
        owner = name[len(_PROBE_TMP_PREFIX) :].split("-", 1)[0]
        if not owner.isdigit() or _pid_alive(int(owner)):
            continue
        # The PID in the NAME is a hint, not proof of ownership, and trusting
        # it alone deleted a RUNNING mongod's dbpath (2026-09-22): the name had
        # been hand-rolled with a shell's `$$` rather than the server's PID, so
        # the sweep saw a long-dead shell, reaped the directory, and mongod
        # died on a fatal WiredTiger assertion -- "log pre-alloc server error
        # ... the process must exit and restart". Ask the STORE whether it is
        # in use before believing the name.
        if _wt_home_in_use(path):
            continue
        try:
            freed += _dir_size(path)
            shutil.rmtree(path, ignore_errors=True)
            reaped += 1
        except OSError:
            continue
    return (reaped, freed)


#: How long a probe store must sit untouched before a dead pid is believed.
#:
#: The pid in the name is not trustworthy on its own -- see `_wt_home_in_use`.
#: A store in use is written to constantly (WiredTiger checkpoints and rolls
#: its log), so a recent mtime is strong evidence something is alive even when
#: the name says otherwise. Half an hour costs nothing: an abandoned store is
#: reclaimed on the next run after that, and the disk problem this whole sweep
#: exists for is measured in days, not minutes.
_PROBE_TMP_GRACE_SECONDS = 1800.0


def _wt_home_in_use(path: str) -> bool:
    """Whether a probe store may still belong to something alive.

    The pid in a store's NAME is written by whoever created the directory, so
    it can simply be wrong -- and when it is, the cost is a live database
    losing its files. That happened on 2026-09-22: a name built from a shell's
    `$$` rather than the server's pid sent this sweep through a running
    mongod's dbpath, and it died on a fatal WiredTiger assertion.

    **Recent modification is the signal, and it is the only one that works
    everywhere.** A store being served is written to constantly, so a fresh
    mtime means hands off. `WiredTiger.lock` looks like the better authority
    and is not: POSIX advisory locks are held per PROCESS, so a check made
    from the process that opened the store reports the file as free -- CI
    failed exactly that way on macOS while Windows passed, because Windows
    locks mandatorily at the handle. Worse, merely opening and closing a
    descriptor to a file this process holds an `fcntl` lock on RELEASES that
    lock, so the "safe" probe can itself break the database it is inspecting.

    On Windows an attempted `os.remove` of the lock IS decisive (measured:
    opening it `r+b` succeeds while WiredTiger holds it and distinguishes
    nothing), so that runs as an extra gate there. It is destructive when it
    answers "free", which is safe only because the caller deletes the whole
    directory immediately afterwards -- do not reuse it as a predicate.

    Ambiguity answers IN USE. A store left behind costs disk; a store deleted
    too early costs data.
    """
    try:
        age = time.time() - os.stat(path).st_mtime
    except OSError:
        return True
    if age < _PROBE_TMP_GRACE_SECONDS:
        return True

    lock = os.path.join(path, "WiredTiger.lock")
    if os.name == "nt" and os.path.isfile(lock):  # pragma: no cover - Windows
        try:
            os.remove(lock)
        except OSError:
            return True
    return False


def _pytest_tmp_owner_alive(path: str) -> bool:
    """Whether the pytest run that owns ``path`` is still running.

    No lock file means nobody claimed it (or the owner exited cleanly and
    removed it) — garbage either way. A lock whose PID no longer exists is
    stale. Anything unreadable is treated as ALIVE: refusing to delete is the
    safe direction when the evidence is unclear.
    """
    lock = os.path.join(path, ".lock")
    try:
        with open(lock) as fh:
            pid = int(fh.read().strip())
    except FileNotFoundError:
        return False
    except (OSError, ValueError):
        return True
    return _pid_alive(pid)


def _pid_alive(pid: int) -> bool:
    """Whether ``pid`` names a running process — cross-platform.

    POSIX uses the ``kill(pid, 0)`` idiom. Windows ``os.kill`` rejects signal 0
    with ``OSError [WinError 87]`` regardless of whether the pid exists (a plain
    ``OSError``, so the POSIX handlers below never saw it — this reaped nothing
    and failed the stale-lock test on Windows), so query the process handle
    instead. An unreadable / ambiguous result is treated as ALIVE — refusing to
    delete is the safe direction."""
    if os.name == "nt":  # pragma: no cover - exercised only on Windows CI
        import ctypes

        PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
        STILL_ACTIVE = 259
        kernel32 = ctypes.windll.kernel32
        handle = kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
        if not handle:
            return False  # no such process (or access denied → treat as gone)
        try:
            code = ctypes.c_ulong()
            if kernel32.GetExitCodeProcess(handle, ctypes.byref(code)):
                return code.value == STILL_ACTIVE
            return True
        finally:
            kernel32.CloseHandle(handle)
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True  # exists but owned by someone else
    return True


def _dir_size(path: str) -> int:
    total = 0
    for dirpath, _dirnames, filenames in os.walk(path, onerror=lambda _e: None):
        for name in filenames:
            try:
                total += os.lstat(os.path.join(dirpath, name)).st_size
            except OSError:
                continue
    return total


@task(
    name="py-gate",
    help={
        "perf": "Also run the perf-regression gate after the suite (default False).",
        "deselect": (
            "Comma-separated pytest node ids to deselect from the suite. "
            f"Defaults to the known local-only failure ({_LOCAL_DESELECT}); pass "
            '--deselect "" to re-include everything.'
        ),
    },
)
def py_gate(c: Context, perf: bool = True, deselect: str = _LOCAL_DESELECT) -> None:
    """Full pre-commit gate for Python-server changes, in one task.

    Runs, in order: ``ruff check`` + ``ruff format --check`` (via ``lint``); the
    full pytest suite (with the known local-only failure deselected by default);
    and — when ``--perf`` — the perf-regression gate. This is the Python-server
    counterpart to ``rust-gate``; it must be green before committing
    ``src/secantus/**`` work.
    """
    steps = 3 if perf else 2
    # ``==> [k/N] label`` phase markers: drive the Ops Board progress stepper
    # (secantus.opsboard.progress) and give a clear CLI banner per sub-step.
    print(f"==> [1/{steps}] Lint", flush=True)
    lint(c)
    print(f"==> [2/{steps}] Tests", flush=True)
    cmd = "uv run python -m pytest -q"
    for nodeid in (d for d in deselect.split(",") if d.strip()):
        cmd += f" --deselect {shlex.quote(nodeid.strip())}"
    c.run(cmd, pty=PTY)
    if perf:
        print(f"==> [3/{steps}] Perf", flush=True)
        perf_task(c)


@task(
    name="py-ship",
    help={
        "message": "Commit message (include the Co-Authored-By trailer yourself).",
        "push": "Push HEAD:main after committing (default True; --no-push to stop at commit).",
        "perf": "Run the perf gate as part of py-gate (default True).",
        "deselect": "Passed through to py-gate (see its help).",
    },
)
def py_ship(
    c: Context,
    message: str,
    push: bool = True,
    perf: bool = True,
    deselect: str = _LOCAL_DESELECT,
) -> None:
    """One-shot: run the full Python-server gate, then commit and push HEAD:main.

    Runs ``py-gate`` (aborts the ship if anything is red), stages all tracked
    modifications **except** the vendored submodules and untracked gauge
    artefacts, commits with ``--message``, and — unless ``--no-push`` — pushes the
    current branch to ``main`` (this repo's feature-branch → main flow). New
    (untracked) files must be ``git add``-ed first. Mirrors ``rust-ship``.
    """
    py_gate(c, perf=perf, deselect=deselect)
    c.run(
        "git add -A -- . "
        "':(exclude)vendor' "
        "':(exclude)secantus-data' "
        "':(exclude,glob)docs/validation-report-*-rust-server.md'",
        pty=PTY,
    )
    c.run(f"git commit -m {shlex.quote(message)}", pty=PTY)
    if push:
        c.run("git push origin HEAD:main", pty=PTY)