scripts/test_hotpath.py

#!/usr/bin/env python3
"""
scripts/test_hotpath.py — the v1.71 send-path fixes, asserted.

    venv/bin/python scripts/test_hotpath.py

Every check here corresponds to a bug that was live in production and that
no existing test could have caught, because each one failed SILENTLY:

  • the signal announce was awaited on the path between finding an edge and
    acting on it, contradicting its own comment
  • no gate anywhere compared a quote's age against a budget
  • quote age, once measured, must not be booked as a pipeline STAGE —
    it overlaps the trace instead of adding to it, and would corrupt
    unaccounted_ms()
  • rate_governor.interval_secs was CALLED as a method though it is a
    property, raising TypeError into a bare `except: pass`
  • `tip` and `diversify` were declared in STAGES and never recorded
  • scanner and executor disagreed about the floor on the same env var
  • a new CSV column must not throw away the existing trace history

Dependency-free (no pytest) and runs in about a second, so it can go in
scripts/check.sh and be run from a phone — same contract as the other
test_*.py files here.
"""

from __future__ import annotations

import ast
import inspect
import os
import sys
import tempfile
import time
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

FAILURES: list[str] = []
CHECKS = 0


def check(label: str, condition: bool, detail: str = "") -> None:
    global CHECKS
    CHECKS += 1
    if condition:
        print(f"  \033[32m✓\033[0m {label}")
    else:
        FAILURES.append(f"{label}{f' — {detail}' if detail else ''}")
        print(f"  \033[31m✗\033[0m {label}" + (f" — {detail}" if detail else ""))


ROOT = Path(__file__).resolve().parent.parent


def code_of(rel: str) -> str:
    """Source with comment lines stripped.

    These checks assert on what the code DOES, and this file's fixes are
    heavily commented with the very strings being searched for — the first
    run of this suite failed on its own explanatory comment quoting the bug
    it had just fixed. Comments describe history; only code is behaviour.
    """
    out = []
    for line in (ROOT / rel).read_text(encoding="utf-8").splitlines():
        stripped = line.lstrip()
        if stripped.startswith("#"):
            continue
        out.append(line.split("  # ")[0] if "  # " in line else line)
    return "\n".join(out)


# ── 1. the announce must not be awaited on the hot path ──────────────────
def test_announce_not_awaited() -> None:
    print("\n\033[1m1. the signal announce is off the hot path\033[0m")
    src = code_of("modules/solana_arb.py")

    check("_fire_and_forget helper exists", "def _fire_and_forget(" in src)

    # The precise regression: `await self._announce(` anywhere in the file.
    # Both call sites are now scheduled, so this substring must be gone.
    check(
        "no `await self._announce(` remains",
        "await self._announce(" not in src,
        "an awaited announce is a Telegram round trip between signal and send",
    )
    check(
        "signal announce is scheduled via _fire_and_forget",
        "_fire_and_forget(" in src and "_format_signal_found(opp)" in src,
    )

    # A strong reference must be held, or asyncio may collect the task
    # mid-flight and the message is silently never sent.
    check("background tasks are strongly referenced", "_BG_ANNOUNCES" in src)


# ── 2. the quote-age gate ────────────────────────────────────────────────
def test_quote_age_gate() -> None:
    print("\n\033[1m2. a stale signal is refused before anything is built\033[0m")
    src = code_of("modules/solana_executor.py")

    check("SOLANA_MAX_QUOTE_AGE_MS is read", "SOLANA_MAX_QUOTE_AGE_MS" in src)
    check("the gate exists", "_MAX_QUOTE_AGE_MS > 0 and quote_age_ms > _MAX_QUOTE_AGE_MS" in src)
    check("it aborts with a named outcome", 'declined:quote_too_old' in src)

    # THE ORDERING CLAIM — and the whole point of the gate. It must sit
    # before the work it exists to avoid paying for. Compare source offsets:
    # everything expensive has to come after it.
    gate_at = src.find("declined:quote_too_old")
    check("gate is present to locate", gate_at > 0)
    if gate_at > 0:
        for stage in ("swap_ix", "blockhash", '"build"', '"sim"', '"send"'):
            first_after = src.find(f'trace.stage({stage})', gate_at)
            first_any = src.find(f'trace.stage({stage})')
            check(
                f"  ...before the `{stage.strip(chr(34))}` stage",
                first_any == -1 or first_after == first_any,
                "an expired quote must not reach this stage",
            )

    # 400ms default == one Solana slot. Not arbitrary; assert it stays.
    from modules import solana_executor as ex

    check(
        "default budget is one 400ms slot",
        abs(ex._MAX_QUOTE_AGE_MS - 400.0) < 1e-9,
        f"got {ex._MAX_QUOTE_AGE_MS}",
    )


# ── 3. quote age must not corrupt unaccounted_ms ─────────────────────────
def test_quote_age_is_not_a_stage() -> None:
    print("\n\033[1m3. quote age is a field, not a stage\033[0m")
    from modules.pipeline_trace import STAGES, PipelineTrace

    check("quote_age is NOT in STAGES", "quote_age" not in STAGES and
          "quote_age_ms" not in STAGES,
          "a stage is subtracted from wall clock; quote age overlaps it")

    t = PipelineTrace(route="USDC->SOL->USDC")
    check("PipelineTrace has a quote_age_ms field", hasattr(t, "quote_age_ms"))

    # The real invariant: booking a large quote age must not eat the
    # unaccounted figure, which is the only signal that instrumentation is
    # incomplete.
    with t.stage("quote1"):
        time.sleep(0.02)
    t.quote_age_ms = 5_000.0          # absurdly stale, far exceeds wall clock
    unacc = t.unaccounted_ms()
    check(
        "a 5,000ms quote age does not zero unaccounted_ms",
        unacc > 0.0,
        f"unaccounted collapsed to {unacc:.1f}ms — quote age is being subtracted",
    )
    check(
        "unaccounted_ms stays below total",
        unacc <= t.total_ms() + 1e-6,
    )


# ── 4. the governor property/method bug ──────────────────────────────────
def test_governor_property_not_called() -> None:
    print("\n\033[1m4. rate governor is actually reachable from the send path\033[0m")
    from modules.rate_governor import get_governor

    gov = get_governor("jupiter")
    check(
        "interval_secs is a property, not a method",
        isinstance(gov.interval_secs, float),
        f"got {type(gov.interval_secs).__name__}",
    )

    # Calling it is the bug. Prove it still raises, so the assertion below
    # is meaningful rather than tautological.
    raised = False
    try:
        gov.interval_secs()          # type: ignore[operator]
    except TypeError:
        raised = True
    check("calling it raises TypeError (the original bug)", raised)

    src = code_of("modules/solana_executor.py")
    check(
        "solana_executor does NOT call it",
        ".interval_secs()" not in src,
        "TypeError swallowed by a bare except == governor silently unused",
    )
    check("solana_executor reads it as a property", ".interval_secs" in src)


# ── 5. declared stages are actually recorded ─────────────────────────────
def test_declared_stages_are_recorded() -> None:
    print("\n\033[1m5. every declared stage is written by someone\033[0m")
    from modules.pipeline_trace import STAGES

    src = code_of("modules/solana_executor.py")

    # `pace` is recorded via trace.add(), not trace.stage() — accept either.
    for stage in STAGES:
        recorded = (f'trace.stage("{stage}")' in src) or (f'trace.add("{stage}"' in src)
        check(f"`{stage}` is recorded", recorded,
              "declared in STAGES but never written — lands in unaccounted_ms")


# ── 6. the two files agree about one env var ─────────────────────────────
def test_pacing_floor_agrees() -> None:
    print("\n\033[1m6. scanner and executor agree on the pacing floor\033[0m")
    from modules import solana_arb, solana_executor

    check(
        "same effective SOLANA_JUPITER_MIN_INTERVAL_SECS",
        abs(solana_arb._JUPITER_MIN_CALL_INTERVAL_SECS
            - solana_executor._JUPITER_MIN_CALL_INTERVAL_SECS) < 1e-9,
        f"arb={solana_arb._JUPITER_MIN_CALL_INTERVAL_SECS} "
        f"executor={solana_executor._JUPITER_MIN_CALL_INTERVAL_SECS}",
    )
    check(
        "executor honours the 2.0s floor",
        solana_executor._JUPITER_MIN_CALL_INTERVAL_SECS >= 2.0
        or solana_executor._JUPITER_ALLOW_FAST_PACING,
    )


# ── 7. the tip bid no longer awaits the network ──────────────────────────
def test_tip_bid_is_non_blocking() -> None:
    print("\n\033[1m7. the tip bid does not await an HTTP call\033[0m")
    from modules.jito_tip_engine import JitoTipEngine

    check("decide_fast exists", hasattr(JitoTipEngine, "decide_fast"))
    check(
        "decide_fast is NOT a coroutine",
        not inspect.iscoroutinefunction(JitoTipEngine.decide_fast),
        "a synchronous bid cannot wait on bundles.jito.wtf",
    )

    src = code_of("modules/solana_executor.py")
    check(
        "executor no longer awaits decide_live",
        "await _get_tip_engine().decide_live(" not in src,
    )
    check("executor uses decide_fast", "decide_fast(" in src)
    check("the tip bid is timed", 'trace.stage("tip")' in src)


# ── 8. the CSV schema change keeps old history ───────────────────────────
def test_trace_history_survives_schema_change() -> None:
    print("\n\033[1m8. adding a column did not throw away the history\033[0m")
    from modules.pipeline_trace import _COLUMNS, _COLUMNS_LEGACY

    check(
        "legacy column set is exactly one narrower",
        len(_COLUMNS) == len(_COLUMNS_LEGACY) + 1,
    )
    check("quote_age_ms is the added column", "quote_age_ms" in _COLUMNS
          and "quote_age_ms" not in _COLUMNS_LEGACY)

    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "legacy.csv"
        n_stage_cols = len(_COLUMNS_LEGACY) - 6
        path.write_text(
            ",".join(_COLUMNS_LEGACY) + "\n"
            + "2026-08-01T10:00:00+00:00,USDC->SOL->USDC,sent,912.4,120.0,350000,"
            + ",".join(["5.0"] * n_stage_cols) + "\n",
            encoding="utf-8",
        )
        os.environ["PIPELINE_TRACE_PATH"] = str(path)
        import importlib

        from modules import pipeline_trace as pt
        importlib.reload(pt)
        log = pt.PipelineLog()
        check("an old-format row still loads", len(log._rows) == 1,
              f"loaded {len(log._rows)} rows — history would be lost")
        if log._rows:
            row = log._rows[0]
            check("its data is intact", row["route"] == "USDC->SOL->USDC"
                  and abs(float(row["total_ms"]) - 912.4) < 1e-6)
            check(
                "missing quote_age reads as absent, not as zero",
                row.get("quote_age_ms") == "",
                "0.0 would claim it was measured and instantaneous",
            )
        os.environ.pop("PIPELINE_TRACE_PATH", None)
        importlib.reload(pt)


# ── 9. an edge that cannot pay the auction is refused early ──────────────
def test_auction_affordability_gate() -> None:
    print("\n\033[1m9. an edge that cannot pay to land is refused before re-quoting\033[0m")
    import types

    os.environ["SOLANA_JITO_ENABLED"] = "true"
    os.environ["SOLANA_MIN_REAL_NET_MARGIN"] = "0.35"

    import modules.solana_arb as sa
    from modules import tip_memory as tm

    eng = sa.SolanaArbEngine.__new__(sa.SolanaArbEngine)
    eng._last_sol_usd_price = 73.0             # from the operator's report

    def opp(net_usd: float) -> Any:
        return types.SimpleNamespace(net_profit_usd=net_usd)

    original = tm.remembered_floor
    try:
        # 0.0032 SOL — the p99 competitive floor the operator was shown.
        tm.remembered_floor = lambda max_age_secs=21600.0: (int(0.0032 * 1e9), 1800.0)

        # THE REGRESSION CASE: the actual best signal of 2026-08-04. It was
        # re-quoted, priced and declined 106 times. It must now not get that far.
        check(
            "the real +$0.0614 signal is declined",
            bool(eng._unaffordable_auction_reason(opp(0.0614))),
            "tip alone is ~$0.23 — landing this loses money",
        )
        check(
            "an edge below tip+margin is declined",
            bool(eng._unaffordable_auction_reason(opp(0.58))),
        )
        check(
            "an edge above tip+margin passes",
            not eng._unaffordable_auction_reason(opp(1.50)),
            "the gate must only ever decline, never over-decline",
        )

        # Rule 2: unknown must never collapse into free.
        tm.remembered_floor = lambda max_age_secs=21600.0: (0, 0.0)
        check(
            "an UNKNOWN floor abstains rather than assuming zero",
            not eng._unaffordable_auction_reason(opp(0.0614)),
            "treating unknown as free makes the safest state the most permissive",
        )

        # No bundling means no auction, so no tip to clear.
        tm.remembered_floor = lambda max_age_secs=21600.0: (int(0.0032 * 1e9), 1800.0)
        os.environ["SOLANA_JITO_ENABLED"] = "false"
        check(
            "with Jito off the gate abstains",
            not eng._unaffordable_auction_reason(opp(0.0614)),
        )
    finally:
        tm.remembered_floor = original
        os.environ.pop("SOLANA_JITO_ENABLED", None)
        os.environ.pop("SOLANA_MIN_REAL_NET_MARGIN", None)


# ── 10. discover.py screens and walks the same size ──────────────────────
def test_discover_ladder_starts_at_probe() -> None:
    print("\n\033[1m10. discover walks the size it screened at\033[0m")
    import importlib.util

    spec = importlib.util.spec_from_file_location(
        "_discover_probe", ROOT / "scripts" / "discover.py")
    src = (ROOT / "scripts" / "discover.py").read_text(encoding="utf-8")
    ns: dict = {}
    exec(src[src.index("def _build_ladder"):src.index("async def main")], ns)
    build = ns["_build_ladder"]

    check(
        "the screening size is the first rung",
        build([250, 1000, 2500, 5000], 100.0)[0] == 100.0,
        "screening at 100 and starting the ladder at 250 drops every pair "
        "whose edge is real at 100",
    )
    check("existing rungs are preserved",
          build([250, 1000], 100.0) == [100.0, 250.0, 1000.0])
    check("a probe already present is not duplicated",
          build([100, 250], 100.0) == [100.0, 250.0])
    check("rungs come back sorted",
          build([5000, 250, 1000], 100.0) == [100.0, 250.0, 1000.0, 5000.0])

    # The silent-drop bug: a pair that dies must be reported, not skipped.
    check("pairs that die on the ladder are recorded",
          "near_misses" in src and "died_at" in src,
          "a bare `continue` made 'examined and rejected' look like "
          "'nothing to examine'")


# ── 11. the tip is anchored to the floor, not to a share of the edge ─────
def test_tip_is_floor_anchored() -> None:
    print("\n\033[1m11. the bid beats the clearing price, it does not donate the edge\033[0m")
    from modules.jito_tip_engine import JitoTipEngine

    e = JitoTipEngine()
    # The operator's real attempt, at $73/SOL: a $0.0643 edge against a live
    # 1,122-lamport floor. The old formula bid 397,910 — 355x the clearing
    # price — and the margin gate then refused to send, every time, which is
    # why /jito has reported `Bundles: 0/0` for the life of this deployment.
    R, C, floor = 880_821, 85_000, 1_122
    d = e.decide(R, C, floor, min_keep_lamports=int(0.02 / 73.0 * 1e9))

    check("it bids to WIN, not to donate", d.tip_lamports < 10_000,
          f"bid {d.tip_lamports:,} lamports against a {floor:,} floor")
    check("the bid still beats the floor", d.tip_lamports > floor)
    check("and it actually bundles now", d.should_bundle,
          "this exact trade produced 0 bundles under the old formula")

    kept = (R - C - d.tip_lamports) / 1e9 * 73.0
    check("it keeps most of the edge", kept > 0.05,
          f"kept ${kept:.4f} of a $0.0643 edge")

    # γ must remain a real ceiling — a huge floor on a small edge must not
    # produce a bid that loses money.
    d2 = e.decide(20_000, 5_000, 100_000, min_keep_lamports=0)
    check("γ still caps: a small edge is priced out, not overbid",
          not d2.should_bundle and d2.tip_lamports == 0)

    # An unknown floor must not collapse the bid to the minimum.
    d3 = e.decide(R, C, 0, min_keep_lamports=0)
    check("no known floor falls back to share-of-edge",
          d3.tip_lamports > 100_000,
          "bidding the bare minimum into an unknown auction would never land")


# ── 12. learned compute-unit sizing survives a restart ───────────────────
def test_cu_sizing_persists() -> None:
    print("\n\033[1m12. the priority fee is sized to what simulation measured\033[0m")
    src = code_of("modules/solana_executor.py")
    check("CU observations are persisted", "_save_cu_observations" in src
          and "_load_cu_observations" in src,
          "an in-memory learner that needs 5 rare samples never engages")
    check("they are loaded at import", "_load_cu_observations()" in src)
    check("the write is atomic", "tmp.replace(path)" in src,
          "a torn state file would be unreadable and silently reset sizing")


# ── 13. the scoreboard must not depend on machine uptime ────────────────
def test_route_scorer_computes_on_a_fresh_boot() -> None:
    print("\n\033[1m13. route scoring works on a machine that just booted\033[0m")
    from modules.route_scorer import _REFRESH_SECS, _Cache

    fresh = _Cache()
    check(
        "the 'never computed' sentinel is -inf",
        fresh.computed_at == float("-inf"),
        f"got {fresh.computed_at!r}",
    )

    # THE BUG, stated as arithmetic. scores() recomputes when
    # `time.monotonic() - computed_at > _REFRESH_SECS`, and on Linux
    # time.monotonic() is seconds since boot. With a 0.0 sentinel that
    # condition asks "has this machine been up for 3 minutes?" instead of
    # "has the scoreboard ever been computed?" — so for the first
    # _REFRESH_SECS after any reboot, scores() returns {}, focus_set()
    # returns None, and plan_cycle applies neither focus nor pruning.
    for uptime in (0.5, 30.0, 179.0):
        check(
            f"  ...at {uptime:.0f}s uptime it still computes",
            (uptime - fresh.computed_at) > _REFRESH_SECS,
            "an empty scoreboard silently disables all pruning and focus",
        )
    check(
        "the old 0.0 sentinel really was broken (so this test means something)",
        not ((30.0 - 0.0) > _REFRESH_SECS),
    )


# ── 14. a 429 costs a full 10 seconds, everywhere ────────────────────────
def test_rate_limit_backoff_is_ten_seconds() -> None:
    print("\n\033[1m14. a 429 pauses 10s in all three paths\033[0m")
    import asyncio

    # Triton: "you MUST pause all requests from that IP for 10 seconds.
    # Attempting to retry immediately will only result in more 429 errors."
    # Three code paths had their own shorter ladder; all three are asserted
    # here because fixing one and leaving two is the same bug.

    # (1) the executor's Jupiter retry
    from modules import solana_executor as ex

    check("executor has a 429-specific backoff",
          ex._RATE_LIMIT_BACKOFF_SECS >= 10.0,
          f"got {ex._RATE_LIMIT_BACKOFF_SECS}")
    check("and it is longer than the ordinary ladder",
          ex._RATE_LIMIT_BACKOFF_SECS > ex._JUPITER_RETRY_MAX_DELAY_SECS,
          "a 429 must not share the timeout ladder")

    class _429(Exception):
        pass

    check("a 429 is recognised", ex._is_rate_limit_error(Exception("HTTP 429")))
    check("a timeout is NOT treated as a 429",
          not ex._is_rate_limit_error(Exception("read timeout")),
          "paying 10s for a network blip costs more signals than it saves")

    # (2) transport.py's retry ladder
    from modules.transport import Transport, TransportConfig

    t = Transport.__new__(Transport)
    t.config = TransportConfig()
    check("transport pauses >=10s on 429", t._backoff_delay(1, 429) >= 10.0,
          f"got {t._backoff_delay(1, 429):.1f}s")
    check("transport stays fast on a 500", t._backoff_delay(1, 500) < 2.0,
          "a server error is not a rate limit")

    # (3) the rate governor — halving the rate is not pausing
    from modules.rate_governor import RateGovernor

    async def _probe() -> float:
        import time as _t

        g = RateGovernor("test-429")
        await g.wait()
        g.rate_limited("simulated")
        t0 = _t.monotonic()
        await g.wait()
        return _t.monotonic() - t0

    waited = asyncio.run(_probe())
    check("governor serves out the full pause", waited >= 10.0,
          f"waited {waited:.1f}s — halving the rate alone still releases "
          f"the next call in ~2s, which re-arms the limit")


def main() -> int:
    print("\n\033[1m══════ v1.71 send-path regressions ══════\033[0m")
    test_announce_not_awaited()
    test_quote_age_gate()
    test_quote_age_is_not_a_stage()
    test_governor_property_not_called()
    test_declared_stages_are_recorded()
    test_pacing_floor_agrees()
    test_tip_bid_is_non_blocking()
    test_trace_history_survives_schema_change()
    test_auction_affordability_gate()
    test_discover_ladder_starts_at_probe()
    test_tip_is_floor_anchored()
    test_cu_sizing_persists()
    test_route_scorer_computes_on_a_fresh_boot()
    test_rate_limit_backoff_is_ten_seconds()

    print()
    if FAILURES:
        print(f"\033[31m❌ {len(FAILURES)} of {CHECKS} checks FAILED\033[0m")
        for f in FAILURES:
            print(f"   • {f}")
        print()
        return 1
    print(f"\033[32m✅ all {CHECKS} checks passed\033[0m\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())