degenbot.operator¶

Operator control surface for a live bot.

A Unix-domain-socket command channel (”degenbot.operator.operator_channel”) that lets an operator steer a running bot — add a specific path or trigger a bounded on-demand discovery — without touching its process. The host runs an OperatorServer; the degenbot path add / degenbot path discover CLI writes commands to it.

Submodules¶

Package Contents¶

degenbot.operator.OperatorHandler¶
class degenbot.operator.OperatorServer(handler: OperatorHandler, *, socket_path: str, request_timeout: float = 60.0)¶

A Unix-domain-socket command server for a live bot.

Run serve() as an asyncio task (e.g. a background task on the registration loop). It accepts one or more concurrent client connections, reads one JSON request line, routes it to the wrapped handler, and writes one JSON response line. The handler is wrapped by wrap_handler() so a failing command replies with {"ok": false} instead of raising into the host.

Parameters:

  • handler – async (op, payload) -> dict (see OperatorHandler).

  • socket_path – filesystem path for the Unix domain socket.

  • request_timeout – seconds to wait for a request line before dropping the connection (default 60).

async serve() → None¶

Start the socket server and accept connections until closed.

Runs forever (cancellable); the caller starts it with asyncio.create_task and cancels it on shutdown. The serve_forever loop runs as its own future so close() can cancel it (without that, wait_closed in close would block on the still-pending loop).

async wait_ready(timeout_s: float = 5.0) → None¶

Wait until serve() has bound the socket and is listening.

Parameters:

timeout_s – seconds to wait before raising TimeoutError.

async close() → None¶

Stop accepting connections and remove the socket file.

Self-sufficient: cancels the in-flight serve_forever loop (so wait_closed() cannot block on it) then closes the server and unlinks the socket. Safe to call whether serve() is still running, was cancelled by the host, or is running on another thread’s event loop.

class degenbot.operator.StepSpec¶

A hop descriptor in the discovery-item shape the pipeline consumes.

type¶

The typed pool family for the hop (V2/V3/V4).

address¶

The pool address (0x + 40 hex).

hash¶

The V4 pool id (0x + 64 hex) when type is PoolKind.V4.

type: degenbot.pathfinding.PoolKind¶
address: str¶
hash: object | None = None¶
async degenbot.operator.send_command(socket_path: str, op: str, payload: dict[str, Any]) → dict[str, Any]¶

Connect to a running OperatorServer and send one command.

Parameters:

  • socket_path – the Unix domain socket path the server listens on.

  • op – command op (add_path / discover).

  • payload – the command payload.

Returns:

The server’s response dict ({"ok": bool, "detail"/"error": ...}).

Raises:

RuntimeError – if the server sends no response line.

degenbot.operator.step_from_wire(step: dict[str, Any]) → StepSpec¶

Translate a wire steps entry into a StepSpec.

Parameters:

step – {"family": "V2|V3|V4", "address": "0x..", "hash": "0x.."?}.

Returns:

A StepSpec mapping family to its typed pool kind.

Raises:

ValueError – if family is not V2/V3/V4 or address is absent.