degenbot.runner.bot_runner

Settlement-arbitrage runtime driver facade (BotRunner).

BotRunner is the Python-companion cockpit over the Rust-owned engine: it owns the config and the lifecycle, while the coordination state itself (the three actors bot/engine_registry/async_w3 + the Dispatcher + the block clock) is owned by the ONE _SessionState built in start(). It is the one place that enforces the phase ordering the engine’s state machine requires.

start(): subscribe -> stream snapshots -> backfill -> verify config

(EngineRegistry.start, stops at Backfilled, pre-resume)

run(): the run ritual (_run_ritual) drives consumer-attach ->

watch-attach -> resume() -> registration -> main loop

The driver is stays-python (asyncio loop, SIGINT, deployment policy): it controls the Rust engine but owns no pool state (ADR-003: Bot is the single state owner; ADR-006: Bot is the per-chain orchestrator). Sequencing supersedure (ADR-050, 2026-09-14): the start/run sequencing described here is Rust-owned in EngineDriver; only asyncio/SIGINT/ deployment policy stays stays-python — see docs/architecture/rust-settlement-bot-parity.md. It delegates path registration to build_paths and the main loop to consume.

Testability seams (mirrors EngineRegistry’s engine= seam): bot, engine_registry, async_w3, snapshots, path_builder, and consumer are injectable.

Module Contents

exception degenbot.runner.bot_runner.ActivationGateRefused(refusal: ValueError)

Bases: RuntimeError

Strategy readiness (the activation gate) refused a live session.

The gate refuses an activated facet that fails readiness, so a live boot either aborts here or degrades a broadcast to the public mempool; the gate’s ValueError text becomes the message so the operator keeps the gate’s own diagnostic unchanged.

exception degenbot.runner.bot_runner.PhaseError

Bases: RuntimeError

Cockpit phase violation: a lifecycle method ran in the wrong phase.

The session phase machine is New -> Started -> Running -> Closed: start() builds once from New and is an idempotent no-op on Started re-entry (Running/Closed re-entry raises); run() requires Started; enqueue_path / trigger_discovery require Running; shutdown() stays deliberately any-phase and idempotent (the SIGINT teardown ordering depends on it).

The transition table itself is the Rust host’s SessionPhase (strategy_host.rs), read through degenbot._ffi.session_phase_next; this exception is the Python surface of its refusals.

class degenbot.runner.bot_runner.InjectedActors

Test/DI actor overrides for BotRunner (None = build from cfg).

bot: degenbot.Bot | None = None

engine_registry: degenbot.arbitrage.engine_registry.EngineRegistry | None = None

async_w3: degenbot.provider.AsyncAlloyProvider | None = None

snapshots: tuple[Any, Any, Any, Any] | None = None

path_builder: Any = None

consumer: Any = None

pipeline_factory: collections.abc.Callable[[_SessionState], Any] | None = None

relay_posture: degenbot.runner._relay_posture.RelayPosture | None = None

settlement_arm: bool | None = None

readiness: collections.abc.Callable[[], degenbot.strategy.StrategyReadinessView] | None = None

settlement_endpoints: collections.abc.Callable[[], list[str]] | None = None

stop_engine: collections.abc.Callable[[], None] | None = None

scheduler: collections.abc.Callable[[collections.abc.Coroutine[Any, Any, None]], asyncio.Task[Any]]

class degenbot.runner.bot_runner.BotRunner(cfg: degenbot.runner.config.ArbitrageConfig, *, actors: InjectedActors | None = None, install_sigint: bool = True)

Orchestrator that collapses the settlement-arbitrage startup ritual behind one facade.

Owns the config and the lifecycle; the coordination state itself (the three actors bot/engine_registry/async_w3, the Dispatcher, the block clock, the sim context, the pipelines) lives on ONE _SessionState built in start() — the runner’s same-named attributes are a facade over that owner, not mirrors. The runner is the ONE place that enforces the phase ordering the engine’s state machine requires:

start(): subscribe → stream snapshots → backfill → verify config

(EngineRegistry.start, stops at Backfilled, pre-resume)

run(): the run ritual (_run_ritual.RunRitual) owns the startup

ordering — attach consumer → watch → resume() → registration → main loop — as a state machine; the cross-task fail-fast channel surfaces a fatal registration error.

Usage (production):

cfg = ArbitrageConfig.build(live=not dry_run, permutation=args.permutation)
async with BotRunner(cfg) as session:
    await session.run()

In production run() schedules discovery+registration as a background task (through the injected scheduler) and enters the main loop immediately; the state-trim runs on registration completion (inside the scheduled hand-off), not on the main-loop entry path. A fatal verification error still crashes loudly through the cross-task channel. The hot loop keeps only engine_registry + async_w3 + dispatcher once trimmed — the Python pool/token caches are scaffolding once the Rust engine owns canonical state.

Testability seams (mirrors EngineRegistry’s engine= seam): bot, engine_registry, async_w3, snapshots, path_builder, consumer, and the registration scheduler are injectable. When injected, start()/run() orchestrate the fakes and the phase ordering is verifiable offline; when None (production), the actors are built from cfg and the real module functions are called.

cfg

property bot: degenbot.Bot | None

The session’s Python-companion bot (None once the trim dropped it).

Before start(): the injected seam (a write is an injection).

property engine_registry: degenbot.arbitrage.engine_registry.EngineRegistry | None

the injected seam.

Type:

The session’s engine registry. Before start()

property async_w3: degenbot.provider.AsyncAlloyProvider | None

the injected seam.

Type:

The session’s dispatch-path provider. Before start()

property dispatcher: degenbot.dispatch.Dispatcher | None

None (no session yet).

Type:

The session’s dispatcher. Before start()

property session: _SessionState | None

The session owner (None only before start()).

property session_watch: degenbot.runner._session_watch.SessionWatch

The session watch the ritual attaches the run’s members to.

property consumer: Any

The injected consumer (None = the production block loop).

property readiness: degenbot.strategy.StrategyReadinessView | None

The readiness view resolved once at the start() boundary.

property settlement_active: bool

The resolved settlement-arm disposition (read by the ritual).

property path_builder: Any

The injected path builder (None = the real build_paths).

property scheduler: collections.abc.Callable[[collections.abc.Coroutine[Any, Any, None]], asyncio.Task[Any]]

The registration hand-off scheduler (data; production create_task).

property trim: collections.abc.Callable[..., None]

The python-state trim (the registration hand-off’s completion duty).

property pump_finished_watchdog: collections.abc.Callable[..., collections.abc.Coroutine[Any, Any, None]]

The pump-finished watchdog factory (the watch’s always-on member).

async start() → BotRunner

Build the actors, fetch block state, load snapshots, run engine_registry.start().

Stops at Backfilled — BEFORE resume(). Zero result batches emit during this window (the pump isn’t running), so run() can attach the consumer in the gap before resume() without a stale-backlog window. Idempotent via the phase alone: re-entry once Started is a no-op; Running/Closed re-entry raises PhaseError.

Returns:

The started runner, stopped at Backfilled.

Raises:

RuntimeError – If the latest-block fetch fails at session start.

async run() → degenbot.runner._session_watch.SessionEndVerdict

Run the cockpit main loop until the consumer task ends.

Requires the Started phase — the session phase machine (PhaseError, delegating to the Rust host’s SessionPhase table) owns the lifecycle gate. The startup ordering itself is the run ritual’s (_run_ritual): this method is the thin phase-gated drive over that machine.

Returns:

The session’s end verdict (the session watch’s ranking over how the run ended). Consuming it is optional — callers that ignore it behave exactly as before this return existed.

async enqueue_path(path_steps: Any, directions: list[bool] | None = None) → None

Add ONE specific path at any time (the operator surface).

Delegates to the session’s live PathRegistrationPipeline (created by the run ritual); path_steps + optional directions are the same shapes as PathRegistrationPipeline.enqueue_path(). The path is built via the retained ConstructionContext (Rust PoolBuilder), registered + verified, released to Live, and registered — without disturbing the pump’s update/solve/dispatch.

Raises:

RuntimeError – if no live pipeline exists (injected fake builders have no construction surface, or run() has not run).

async trigger_discovery(*, bound: int | None = None) → int

Trigger a bounded one-shot discovery sweep (on-demand trigger).

Delegates to the session’s live pipeline.

Returns:

The number of paths processed by the sweep.

Raises:

RuntimeError – if no live pipeline exists (injected fake builders, or run() has not run).

async shutdown() → None

Signal the Rust core to stop the pump (best-effort).

Safe to call at any point in the lifecycle — before start() finished (engine_registry may be None), after run() exited, or from a SIGINT/KeyboardInterrupt handler. Mirrors the Rust stop() contract: idempotent, sets the shutdown flag + aborts the pump task so the WS stream’s combined.next().await unblocks immediately (60s cold-shutdown otherwise). Any exception is swallowed and logged so a partial-startup teardown can’t mask the original in-flight exception.

This is the one place that closes the Rust core’s pump — the KeyboardInterrupt-exits-slowly bug was the pump task (spawned on the shared tokio runtime, decoupled from the asyncio loop) blocking on a silent WS subscription, which asyncio.run’s teardown did not reach until the OS closed the socket.