feat(trading): zero-RPC extreme-fast buys and listener race hardening (#170) (#183)

Extreme-fast mode submits with no RPC calls between detection and buy
when the listener parsed the on-chain CreateEvent (geyser/logs/blocks):
TokenInfo.state_from_event carries the canonical creator, mayhem/cashback
flags and quote_mint, so the pre-buy curve refresh is skipped. The geyser
listener now delegates to the event-first parser instead of inlining
instruction decoding, which also catches create_v2 transactions that omit
the trailing is_cashback_enabled byte.

Listeners without event data (pumpportal) keep the refresh, hardened per
issue #170: bonding curve derived from the mint instead of the payload's
bondingCurveKey, curve and mint read in one slot-consistent
getMultipleAccounts (correcting the guessed token program), and an
unreadable curve now skips the token within trade.curve_refresh_budget
instead of submitting a buy built from guessed accounts.

Machine-checked by verify_extreme_fast_zero_rpc.py and
verify_pumpportal_buy_path.py; validated live via simulate_bot_buy_path
on both geyser and pumpportal paths.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

learning-examples/simulate_bot_buy_path.py

@@ -166,6 +166,7 @@ async def main() -> int:
     print(f"quote_mint (from CreateEvent): {token_info.quote_mint}")
     print(f"token program: {token_info.token_program_id}")
     print(f"mayhem={token_info.is_mayhem_mode} cashback={token_info.is_cashback_coin}")
+    print(f"state_from_event={token_info.state_from_event} (True = zero-RPC buy path)")
     print(f"extreme_fast_mode={extreme_fast}\n")
 
     client = SolanaClient(os.environ["SOLANA_NODE_RPC_ENDPOINT"])

learning-examples/verify_extreme_fast_zero_rpc.py

@@ -0,0 +1,432 @@
+"""Verify extreme_fast_mode makes zero RPC calls for event-sourced tokens.
+
+extreme_fast_mode's contract is that nothing sits between detecting a token
+and submitting the buy — no reads, no price fetch. The pump.fun CreateEvent
+carries the canonical creator (instruction args.creator is user-supplied and
+may differ post-2026-04-28), mayhem/cashback flags and quote_mint, so a
+TokenInfo built from it needs no pre-buy curve refresh. PumpPortal payloads
+carry none of that, so they keep the refresh.
+
+Offline machine checks, no network and no funds moved:
+
+  1. The logs parser marks CreateEvent-sourced TokenInfo as state_from_event.
+  2. The instruction parser stays conservative (args.creator not canonical).
+  3. The geyser parser prefers the CreateEvent from meta.log_messages.
+  4. The geyser LISTENER delegates to that parser (it used to inline
+     instruction decoding, bypassing the event path).
+  5. The block parser rides the same CreateEvent logs.
+  6. The pumpportal processor never sets state_from_event.
+  7. An event-sourced buy submits with ZERO curve-manager/RPC calls.
+  8. A pumpportal-sourced buy still refreshes from chain.
+  9. trade.trust_create_event=false forces the refresh even for event data.
+
+Usage:
+    uv run learning-examples/verify_extreme_fast_zero_rpc.py
+"""
+
+import asyncio
+import base64
+import json
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(PROJECT_ROOT / "src"))
+
+from solders.pubkey import Pubkey  # noqa: E402
+from solders.transaction import VersionedTransaction  # noqa: E402
+
+from core.pubkeys import WSOL_MINT, SystemAddresses  # noqa: E402
+from interfaces.core import Platform, TokenInfo  # noqa: E402
+from platforms.pumpfun.address_provider import PumpFunAddressProvider  # noqa: E402
+from platforms.pumpfun.event_parser import PumpFunEventParser  # noqa: E402
+from platforms.pumpfun.pumpportal_processor import (  # noqa: E402
+    PumpFunPumpPortalProcessor,
+)
+from trading import platform_aware  # noqa: E402
+from trading.platform_aware import PlatformAwareBuyer  # noqa: E402
+from utils.idl_manager import get_idl_manager  # noqa: E402
+
+FIXTURE = (
+    PROJECT_ROOT
+    / "learning-examples"
+    / "blocksubscribe-transactions"
+    / "raw_create_tx_from_blocksubscribe.json"
+)
+
+PROVIDER = PumpFunAddressProvider()
+TRADER = Pubkey.from_string("11111111111111111111111111111112")
+
+
+def _event_parser() -> PumpFunEventParser:
+    """Real pump.fun event parser with the vendored IDL, no RPC client needed."""
+    return PumpFunEventParser(
+        idl_parser=get_idl_manager().get_parser(Platform.PUMP_FUN)
+    )
+
+
+def _fixture() -> dict:
+    return json.loads(FIXTURE.read_text())
+
+
+def _event_sourced_token_info() -> TokenInfo:
+    """Parse the fixture's CreateEvent through the real logs parser."""
+    parser = _event_parser()
+    return parser.parse_token_creation_from_logs(
+        _fixture()["meta"]["logMessages"], signature="fixture"
+    )
+
+
+class _StubClient:
+    """Records submissions and account reads; never touches the network."""
+
+    def __init__(self) -> None:
+        self.sent: list = []
+        self.reads = 0
+
+    async def build_and_send_transaction(
+        self, instructions: list, *_args: object, **_kwargs: object
+    ) -> str:
+        self.sent.append(instructions)
+        return "STUB_SIGNATURE"
+
+    async def confirm_transaction(self, _signature: str, **_kwargs: object) -> bool:
+        return False
+
+    async def get_account_info(self, *_args: object, **_kwargs: object) -> None:
+        self.reads += 1
+        raise ValueError("unexpected RPC read in zero-RPC path")  # noqa: TRY003
+
+    async def get_multiple_accounts(self, *_args: object, **_kwargs: object) -> None:
+        self.reads += 1
+        raise ValueError("unexpected RPC read in zero-RPC path")  # noqa: TRY003
+
+
+class _CountingCurveManager:
+    """Counts refresh calls; returns benign state."""
+
+    def __init__(self) -> None:
+        self.calls = 0
+
+    async def get_pool_state_and_token_program(
+        self,
+        _pool: Pubkey,
+        _mint: Pubkey,
+        commitment: str | None = None,  # noqa: ARG002
+    ) -> tuple[dict, Pubkey]:
+        self.calls += 1
+        state = {
+            "creator": str(TRADER),
+            "is_mayhem_mode": False,
+            "is_cashback_coin": False,
+            "quote_mint": WSOL_MINT,
+        }
+        return state, SystemAddresses.TOKEN_2022_PROGRAM
+
+    async def get_pool_state(
+        self,
+        _pool: Pubkey,
+        commitment: str | None = None,  # noqa: ARG002
+    ) -> dict:
+        self.calls += 1
+        return {
+            "creator": str(TRADER),
+            "is_mayhem_mode": False,
+            "is_cashback_coin": False,
+            "quote_mint": WSOL_MINT,
+        }
+
+
+def _stub_implementations(curve_manager: object) -> SimpleNamespace:
+    async def build_buy_instruction(*_args: object, **_kwargs: object) -> list[str]:
+        return ["stub-instruction"]
+
+    instruction_builder = SimpleNamespace(
+        build_buy_instruction=build_buy_instruction,
+        get_required_accounts_for_buy=lambda *_a, **_k: [],
+        get_buy_compute_unit_limit=lambda _override: 100_000,
+    )
+    return SimpleNamespace(
+        address_provider=PROVIDER,
+        instruction_builder=instruction_builder,
+        curve_manager=curve_manager,
+    )
+
+
+def _make_buyer(client: _StubClient, **kwargs: object) -> PlatformAwareBuyer:
+    async def no_fee(_accounts: list) -> None:
+        return None
+
+    fee_manager = SimpleNamespace(calculate_priority_fee=no_fee)
+    return PlatformAwareBuyer(
+        client,
+        SimpleNamespace(pubkey=TRADER, keypair=None),
+        fee_manager,
+        amount=0.0001,
+        slippage=0.3,
+        max_retries=1,
+        extreme_fast_token_amount=20,
+        extreme_fast_mode=True,
+        **kwargs,
+    )
+
+
+def _run_buy(
+    token_info: TokenInfo, curve_manager: object, **buyer_kwargs: object
+) -> tuple[_StubClient, object]:
+    client = _StubClient()
+    buyer = _make_buyer(client, **buyer_kwargs)
+    platform_aware.get_platform_implementations = lambda _p, _c: _stub_implementations(
+        curve_manager
+    )
+    result = asyncio.run(buyer.execute(token_info))
+    return client, result
+
+
+def check_logs_parser_marks_event_state() -> bool:
+    """CreateEvent-sourced TokenInfo carries everything -> flag set."""
+    token_info = _event_sourced_token_info()
+    if token_info is None:
+        print("    fixture logs did not parse into a TokenInfo")
+        return False
+    ok = (
+        getattr(token_info, "state_from_event", False) is True
+        and token_info.quote_mint is not None
+        and token_info.creator is not None
+    )
+    if not ok:
+        print(
+            f"    state_from_event={getattr(token_info, 'state_from_event', None)} "
+            f"quote_mint={token_info.quote_mint} creator={token_info.creator}"
+        )
+    return ok
+
+
+def check_instruction_parser_stays_conservative() -> bool:
+    """args.creator is user-supplied, not canonical -> flag must stay unset."""
+    fixture = _fixture()
+    raw = base64.b64decode(fixture["transaction"][0])
+    tx = VersionedTransaction.from_bytes(raw)
+    msg = tx.message
+    account_keys = [bytes(k) for k in msg.account_keys]
+    parser = _event_parser()
+    for ix in msg.instructions:
+        data = bytes(ix.data)
+        # The fixture's create_v2 omits the trailing is_cashback_enabled
+        # OptionBool (a legal wire form the strict IDL decoder rejects — see
+        # the "decode trailing args defensively" gotcha in CLAUDE.md), so
+        # append the byte to exercise the parser's flag behaviour.
+        if data.startswith(bytes.fromhex("d6904cec5f8b31b4")):
+            data += b"\x00"
+        token_info = parser.parse_token_creation_from_instruction(
+            data, list(ix.accounts), account_keys
+        )
+        if token_info is not None:
+            ok = getattr(token_info, "state_from_event", False) is False
+            if not ok:
+                print("    instruction-sourced TokenInfo must not set the flag")
+            return ok
+    print("    fixture create instruction did not parse")
+    return False
+
+
+def check_geyser_parser_prefers_event_logs() -> bool:
+    """Geyser meta carries log_messages; the CreateEvent there is canonical."""
+    logs = _fixture()["meta"]["logMessages"]
+    stub = SimpleNamespace(
+        transaction=SimpleNamespace(
+            transaction=SimpleNamespace(
+                transaction=SimpleNamespace(
+                    message=SimpleNamespace(instructions=[], account_keys=[])
+                ),
+                meta=SimpleNamespace(log_messages=logs),
+            )
+        )
+    )
+    parser = _event_parser()
+    token_info = parser.parse_token_creation_from_geyser(stub)
+    ok = (
+        token_info is not None
+        and getattr(token_info, "state_from_event", False) is True
+    )
+    if not ok:
+        print(f"    geyser parse returned {token_info}")
+    return ok
+
+
+def check_geyser_listener_delegates_to_parser() -> bool:
+    """The listener must route updates through the event-first parser.
+
+    Live run showed state_from_event=False on a geyser token: the listener
+    inlined instruction parsing and never reached the parser's log-preferring
+    geyser method.
+    """
+    # Imported here: the listener module pulls in grpc, which the other
+    # checks don't need.
+    from monitoring.universal_geyser_listener import (  # noqa: PLC0415
+        UniversalGeyserListener,
+    )
+
+    listener = UniversalGeyserListener(
+        geyser_endpoint="dummy:443",
+        geyser_api_token="dummy",  # noqa: S106 - offline stub, never connects
+        geyser_auth_type="x-token",
+        platforms=[Platform.PUMP_FUN],
+    )
+    logs = _fixture()["meta"]["logMessages"]
+    update = SimpleNamespace(
+        HasField=lambda field: field == "transaction",
+        transaction=SimpleNamespace(
+            transaction=SimpleNamespace(
+                transaction=SimpleNamespace(
+                    message=SimpleNamespace(instructions=[], account_keys=[])
+                ),
+                meta=SimpleNamespace(log_messages=logs),
+            )
+        ),
+    )
+    token_info = asyncio.run(listener._process_update(update))  # noqa: SLF001
+    ok = (
+        token_info is not None
+        and getattr(token_info, "state_from_event", False) is True
+    )
+    if not ok:
+        print(f"    listener _process_update returned {token_info}")
+    return ok
+
+
+def check_block_parser_marks_event_state() -> bool:
+    """The block listener's parse path also rides the CreateEvent logs."""
+    parser = _event_parser()
+    token_info = parser.parse_token_creation_from_block({"transactions": [_fixture()]})
+    ok = (
+        token_info is not None
+        and getattr(token_info, "state_from_event", False) is True
+    )
+    if not ok:
+        print(f"    block parse returned {token_info}")
+    return ok
+
+
+def check_pumpportal_never_sets_flag() -> bool:
+    """PumpPortal payloads carry no curve state -> flag must stay unset."""
+    mint = Pubkey.from_string("So11111111111111111111111111111111111111112")
+    token_info = PumpFunPumpPortalProcessor().process_token_data(
+        {
+            "name": "T",
+            "symbol": "T",
+            "mint": str(mint),
+            "bondingCurveKey": str(PROVIDER.derive_pool_address(mint)),
+            "traderPublicKey": str(TRADER),
+            "uri": "",
+            "pool": "pump",
+        }
+    )
+    ok = (
+        token_info is not None
+        and getattr(token_info, "state_from_event", False) is False
+    )
+    if not ok:
+        print("    pumpportal TokenInfo must not set state_from_event")
+    return ok
+
+
+def check_event_sourced_buy_is_zero_rpc() -> bool:
+    """The killer feature: detection -> submission with no reads at all."""
+    token_info = _event_sourced_token_info()
+    if token_info is None:
+        print("    fixture logs did not parse into a TokenInfo")
+        return False
+    curve_manager = _CountingCurveManager()
+    client, _result = _run_buy(token_info, curve_manager)
+    ok = curve_manager.calls == 0 and client.reads == 0 and len(client.sent) == 1
+    if not ok:
+        print(
+            f"    curve_manager.calls={curve_manager.calls} "
+            f"client.reads={client.reads} submissions={len(client.sent)}"
+        )
+    return ok
+
+
+def check_pumpportal_buy_still_refreshes() -> bool:
+    """Listener-guessed data must still be refreshed from chain."""
+    mint = Pubkey.from_string("So11111111111111111111111111111111111111112")
+    bonding_curve = PROVIDER.derive_pool_address(mint)
+    token_info = TokenInfo(
+        name="T",
+        symbol="T",
+        uri="",
+        mint=mint,
+        platform=Platform.PUMP_FUN,
+        bonding_curve=bonding_curve,
+        associated_bonding_curve=PROVIDER.derive_associated_bonding_curve(
+            mint, bonding_curve, SystemAddresses.TOKEN_2022_PROGRAM
+        ),
+        user=TRADER,
+        creator=TRADER,
+        creator_vault=PROVIDER.derive_creator_vault(TRADER),
+        token_program_id=SystemAddresses.TOKEN_2022_PROGRAM,
+    )
+    curve_manager = _CountingCurveManager()
+    client, _result = _run_buy(token_info, curve_manager)
+    ok = curve_manager.calls >= 1 and len(client.sent) == 1
+    if not ok:
+        print(
+            f"    curve_manager.calls={curve_manager.calls} "
+            f"submissions={len(client.sent)}"
+        )
+    return ok
+
+
+def check_trust_flag_forces_refresh() -> bool:
+    """trust_create_event=false is the escape hatch back to always-refresh."""
+    token_info = _event_sourced_token_info()
+    if token_info is None:
+        print("    fixture logs did not parse into a TokenInfo")
+        return False
+    curve_manager = _CountingCurveManager()
+    _client, _result = _run_buy(token_info, curve_manager, trust_create_event=False)
+    ok = curve_manager.calls >= 1
+    if not ok:
+        print(f"    curve_manager.calls={curve_manager.calls} (expected >=1)")
+    return ok
+
+
+def main() -> int:
+    checks = [
+        ("logs parser marks CreateEvent state", check_logs_parser_marks_event_state),
+        (
+            "instruction parser stays conservative",
+            check_instruction_parser_stays_conservative,
+        ),
+        (
+            "geyser parser prefers CreateEvent logs",
+            check_geyser_parser_prefers_event_logs,
+        ),
+        (
+            "geyser listener delegates to event-first parser",
+            check_geyser_listener_delegates_to_parser,
+        ),
+        ("block parser marks CreateEvent state", check_block_parser_marks_event_state),
+        ("pumpportal never sets state_from_event", check_pumpportal_never_sets_flag),
+        ("event-sourced buy makes zero RPC calls", check_event_sourced_buy_is_zero_rpc),
+        ("pumpportal buy still refreshes", check_pumpportal_buy_still_refreshes),
+        ("trust_create_event=false forces refresh", check_trust_flag_forces_refresh),
+    ]
+    failed = 0
+    for label, check in checks:
+        try:
+            ok = check()
+        except Exception as error:  # noqa: BLE001 - report and continue
+            print(f"FAIL {label}: {type(error).__name__}: {error}")
+            failed += 1
+            continue
+        print(f"{'PASS' if ok else 'FAIL'} {label}")
+        failed += 0 if ok else 1
+    print(f"\n{len(checks) - failed}/{len(checks)} checks passed")
+    return 1 if failed else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

learning-examples/verify_pumpportal_buy_path.py

@@ -0,0 +1,345 @@
+"""Verify the pumpportal-sourced buy path hardening from issue #170.
+
+Offline machine checks, no network and no funds moved:
+
+  A. The pumpportal processor derives bonding_curve from the mint instead of
+     trusting the payload's bondingCurveKey (observed stale once in #170).
+  B. In extreme_fast_mode, a buy is SKIPPED when the curve state cannot be
+     read within the refresh budget, instead of submitting a buy built from
+     listener-guessed defaults (the "racing a doomed buy" failure).
+  C. The curve refresh reads curve + mint in one slot-consistent
+     getMultipleAccounts round trip and corrects token_program_id (pumpportal
+     cannot know it and guesses Token-2022; legacy coins are SPL Token).
+
+Usage:
+    uv run learning-examples/verify_pumpportal_buy_path.py
+"""
+
+import asyncio
+import json
+import struct
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(PROJECT_ROOT / "src"))
+
+from solders.pubkey import Pubkey  # noqa: E402
+
+from core.pubkeys import WSOL_MINT, SystemAddresses  # noqa: E402
+from interfaces.core import Platform, TokenInfo  # noqa: E402
+from platforms.pumpfun.address_provider import PumpFunAddressProvider  # noqa: E402
+from platforms.pumpfun.curve_manager import PumpFunCurveManager  # noqa: E402
+from platforms.pumpfun.pumpportal_processor import (  # noqa: E402
+    PumpFunPumpPortalProcessor,
+)
+from trading import platform_aware  # noqa: E402
+from trading.platform_aware import PlatformAwareBuyer  # noqa: E402
+from utils.idl_manager import get_idl_manager  # noqa: E402
+
+MINT = Pubkey.from_string("So11111111111111111111111111111111111111112")
+TRADER = Pubkey.from_string("11111111111111111111111111111112")
+WRONG_BC = Pubkey.from_string("Vote111111111111111111111111111111111111111")
+
+PROVIDER = PumpFunAddressProvider()
+
+
+def _fabricated_curve_bytes(creator: Pubkey, *, is_mayhem: bool) -> bytes:
+    """Build a 151-byte BondingCurve account image matching the IDL layout."""
+    idl = json.loads((PROJECT_ROOT / "idl" / "pump_fun_idl.json").read_text())
+    disc = next(
+        bytes(a["discriminator"])
+        for a in idl["accounts"]
+        if a["name"] == "BondingCurve"
+    )
+    reserves = struct.pack(
+        "<QQQQQ",
+        1_000_000_000_000,  # virtual_token_reserves
+        30_000_000_000,  # virtual_quote_reserves
+        800_000_000_000,  # real_token_reserves
+        0,  # real_quote_reserves
+        1_000_000_000_000,  # token_total_supply
+    )
+    return (
+        disc
+        + reserves
+        + b"\x00"  # complete
+        + bytes(creator)
+        + (b"\x01" if is_mayhem else b"\x00")  # is_mayhem_mode
+        + b"\x00"  # is_cashback_coin
+        + bytes(32)  # quote_mint = Pubkey::default() (SOL-paired)
+        + bytes(36)  # reserved padding
+    )
+
+
+def _pumpportal_token_info(**overrides: object) -> TokenInfo:
+    """TokenInfo shaped like the pumpportal processor's output."""
+    bonding_curve = PROVIDER.derive_pool_address(MINT)
+    defaults: dict = {
+        "name": "T",
+        "symbol": "T",
+        "uri": "",
+        "mint": MINT,
+        "platform": Platform.PUMP_FUN,
+        "bonding_curve": bonding_curve,
+        "associated_bonding_curve": PROVIDER.derive_associated_bonding_curve(
+            MINT, bonding_curve, SystemAddresses.TOKEN_2022_PROGRAM
+        ),
+        "user": TRADER,
+        "creator": TRADER,
+        "creator_vault": PROVIDER.derive_creator_vault(TRADER),
+        "token_program_id": SystemAddresses.TOKEN_2022_PROGRAM,
+    }
+    defaults.update(overrides)
+    return TokenInfo(**defaults)
+
+
+class _StubClient:
+    """Records submissions; never touches the network."""
+
+    def __init__(self) -> None:
+        self.sent: list = []
+
+    async def build_and_send_transaction(
+        self, instructions: list, *_args: object, **_kwargs: object
+    ) -> str:
+        self.sent.append(instructions)
+        return "STUB_SIGNATURE"
+
+    async def confirm_transaction(self, _signature: str, **_kwargs: object) -> bool:
+        return False
+
+
+def _stub_implementations(curve_manager: object) -> SimpleNamespace:
+    async def build_buy_instruction(*_args: object, **_kwargs: object) -> list[str]:
+        return ["stub-instruction"]
+
+    instruction_builder = SimpleNamespace(
+        build_buy_instruction=build_buy_instruction,
+        get_required_accounts_for_buy=lambda *_a, **_k: [],
+        get_buy_compute_unit_limit=lambda _override: 100_000,
+    )
+    return SimpleNamespace(
+        address_provider=PROVIDER,
+        instruction_builder=instruction_builder,
+        curve_manager=curve_manager,
+    )
+
+
+def _make_buyer(client: _StubClient, **kwargs: float) -> PlatformAwareBuyer:
+    async def no_fee(_accounts: list) -> None:
+        return None
+
+    fee_manager = SimpleNamespace(calculate_priority_fee=no_fee)
+    return PlatformAwareBuyer(
+        client,
+        SimpleNamespace(pubkey=TRADER, keypair=None),
+        fee_manager,
+        amount=0.0001,
+        slippage=0.3,
+        max_retries=1,
+        extreme_fast_token_amount=20,
+        extreme_fast_mode=True,
+        **kwargs,
+    )
+
+
+def check_a_processor_derives_bonding_curve() -> bool:
+    """A: payload bondingCurveKey is not trusted; the PDA is derived."""
+    token_data = {
+        "name": "T",
+        "symbol": "T",
+        "mint": str(MINT),
+        "bondingCurveKey": str(WRONG_BC),  # deliberately stale/wrong
+        "traderPublicKey": str(TRADER),
+        "uri": "",
+        "pool": "pump",
+    }
+    token_info = PumpFunPumpPortalProcessor().process_token_data(token_data)
+    expected = PROVIDER.derive_pool_address(MINT)
+    ok = token_info is not None and token_info.bonding_curve == expected
+    if not ok:
+        got = token_info.bonding_curve if token_info else None
+        print(f"    expected derived BC {expected}, got {got}")
+    return ok
+
+
+def check_b_skips_when_curve_unreadable() -> bool:
+    """B: refresh failure -> buy skipped, nothing submitted."""
+
+    class NeverReadable:
+        async def get_pool_state(
+            self,
+            _pool: Pubkey,
+            commitment: str | None = None,  # noqa: ARG002
+        ) -> dict:
+            raise ValueError("Account not found")  # noqa: TRY003
+
+    client = _StubClient()
+    buyer = _make_buyer(client, curve_refresh_budget=0.3)
+    platform_aware.get_platform_implementations = lambda _p, _c: _stub_implementations(
+        NeverReadable()
+    )
+    result = asyncio.run(buyer.execute(_pumpportal_token_info()))
+    ok = not result.success and not client.sent
+    if not ok:
+        print(f"    success={result.success} submissions={len(client.sent)}")
+    return ok
+
+
+def check_b_still_buys_when_curve_readable() -> bool:
+    """B guard: a readable curve still reaches submission."""
+
+    class Readable:
+        async def get_pool_state(
+            self,
+            _pool: Pubkey,
+            commitment: str | None = None,  # noqa: ARG002
+        ) -> dict:
+            return {
+                "creator": str(TRADER),
+                "is_mayhem_mode": False,
+                "is_cashback_coin": False,
+                "quote_mint": WSOL_MINT,
+            }
+
+    client = _StubClient()
+    buyer = _make_buyer(client, curve_refresh_budget=0.3)
+    platform_aware.get_platform_implementations = lambda _p, _c: _stub_implementations(
+        Readable()
+    )
+    asyncio.run(buyer.execute(_pumpportal_token_info()))
+    ok = len(client.sent) == 1
+    if not ok:
+        print(f"    submissions={len(client.sent)} (expected 1)")
+    return ok
+
+
+def check_c_curve_manager_batch_read() -> bool:
+    """C: curve manager reads curve + mint owner in one batch call."""
+    creator = TRADER
+    curve_bytes = _fabricated_curve_bytes(creator, is_mayhem=True)
+
+    class BatchClient:
+        def __init__(self) -> None:
+            self.batch_calls = 0
+
+        async def get_multiple_accounts(
+            self,
+            pubkeys: list[Pubkey],
+            commitment: str | None = None,  # noqa: ARG002
+        ) -> list[SimpleNamespace]:
+            self.batch_calls += 1
+            if len(pubkeys) != 2:  # noqa: PLR2004
+                raise ValueError("expected [curve, mint]")  # noqa: TRY003
+            return [
+                SimpleNamespace(data=curve_bytes, owner=PROVIDER.program_id),
+                SimpleNamespace(data=b"", owner=SystemAddresses.TOKEN_PROGRAM),
+            ]
+
+    client = BatchClient()
+    manager = PumpFunCurveManager(
+        client, get_idl_manager().get_parser(Platform.PUMP_FUN)
+    )
+    if not hasattr(manager, "get_pool_state_and_token_program"):
+        print("    PumpFunCurveManager.get_pool_state_and_token_program missing")
+        return False
+    state, token_program = asyncio.run(
+        manager.get_pool_state_and_token_program(
+            PROVIDER.derive_pool_address(MINT), MINT, commitment="processed"
+        )
+    )
+    ok = (
+        client.batch_calls == 1
+        and token_program == SystemAddresses.TOKEN_PROGRAM
+        and state.get("is_mayhem_mode") is True
+        and str(state.get("creator")) == str(creator)
+    )
+    if not ok:
+        print(
+            f"    batch_calls={client.batch_calls} token_program={token_program} state={state}"
+        )
+    return ok
+
+
+def check_c_buyer_corrects_token_program() -> bool:
+    """C: buyer applies the batch-read token program and re-derives the ATA."""
+
+    class BatchCurveManager:
+        async def get_pool_state_and_token_program(
+            self,
+            _pool: Pubkey,
+            _mint: Pubkey,
+            commitment: str | None = None,  # noqa: ARG002
+        ) -> tuple[dict, Pubkey]:
+            state = {
+                "creator": str(TRADER),
+                "is_mayhem_mode": False,
+                "is_cashback_coin": False,
+                "quote_mint": WSOL_MINT,
+            }
+            return state, SystemAddresses.TOKEN_PROGRAM
+
+        async def get_pool_state(
+            self,
+            _pool: Pubkey,
+            commitment: str | None = None,  # noqa: ARG002
+        ) -> dict:
+            raise AssertionError("batch method should be preferred")  # noqa: TRY003
+
+    client = _StubClient()
+    buyer = _make_buyer(client, curve_refresh_budget=0.3)
+    platform_aware.get_platform_implementations = lambda _p, _c: _stub_implementations(
+        BatchCurveManager()
+    )
+    token_info = _pumpportal_token_info()
+    asyncio.run(buyer.execute(token_info))
+    expected_ata = PROVIDER.derive_associated_bonding_curve(
+        MINT, token_info.bonding_curve, SystemAddresses.TOKEN_PROGRAM
+    )
+    ok = (
+        token_info.token_program_id == SystemAddresses.TOKEN_PROGRAM
+        and token_info.associated_bonding_curve == expected_ata
+    )
+    if not ok:
+        print(
+            f"    token_program_id={token_info.token_program_id} "
+            f"ata={token_info.associated_bonding_curve} (expected {expected_ata})"
+        )
+    return ok
+
+
+def main() -> int:
+    checks = [
+        (
+            "A: processor derives bonding_curve from mint",
+            check_a_processor_derives_bonding_curve,
+        ),
+        ("B: unreadable curve -> buy skipped", check_b_skips_when_curve_unreadable),
+        ("B: readable curve -> buy proceeds", check_b_still_buys_when_curve_readable),
+        (
+            "C: curve manager batch-reads curve + mint owner",
+            check_c_curve_manager_batch_read,
+        ),
+        (
+            "C: buyer corrects token_program_id and ATA",
+            check_c_buyer_corrects_token_program,
+        ),
+    ]
+    failed = 0
+    for label, check in checks:
+        try:
+            ok = check()
+        except Exception as error:  # noqa: BLE001 - report and continue
+            print(f"FAIL {label}: {type(error).__name__}: {error}")
+            failed += 1
+            continue
+        print(f"{'PASS' if ok else 'FAIL'} {label}")
+        failed += 0 if ok else 1
+    print(f"\n{len(checks) - failed}/{len(checks)} checks passed")
+    return 1 if failed else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

learning-examples/verify_tx_status_checks.py

@@ -144,6 +144,9 @@ async def check_examples_call_a_status_check() -> None:
         Path(__file__).name,
         # stubs confirm_transaction out; never sends a transaction
         "simulate_bot_buy_path.py",
+        # offline checks with a stub client; never sends a transaction
+        "verify_pumpportal_buy_path.py",
+        "verify_extreme_fast_zero_rpc.py",
         # uses the bot's SolanaClient wrapper, which folds meta.err into its
         # return value; the boolean is read at the call site
         "cleanup_accounts.py",