#!/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())