plesty.server.services.agent

The bench agent: runs plesty-server commands for a remote plesty.bench client.

A ZMQ ROUTER on the agent port (5550) speaking the plesty-lib device-server wire. Four request types — describe, exec, fetch, device — documented in plesty.bench.client. exec runs the click application in-process with its output captured, with the agent’s own --home/ -f prepended, so every command the CLI has is remote with nothing to keep in sync. Each request is handled on a worker thread (a start waits for its probe; a describe must not wait behind it); replies go back through the socket thread, since a ZMQ socket belongs to one thread.

device relays one request to a declared device’s own server and hands its reply back. It is the exception to “the CLI is the API”, and deliberately: a control panel is a long-lived session of many small round trips, and exec would run the click application for each one. It is also why the panel is relayed rather than pointed at the device port — a device server binds tcp://* and checks no token, so routing through the agent keeps one authenticated door rather than one per instrument.

Trust model: the bench network, like the device servers — a shared token (--token / PLESTY_BENCH_TOKEN) checked on every exec/fetch/ device, fetch confined to the bench home, device confined to the declared devices this bench is running, no agent/gui through exec.

Attributes

DEFAULT_BIND

TOKEN_ENV

FORBIDDEN

FETCH_LIMIT

DEVICE_TIMEOUT_MS

_STREAMS_LOCK

_SOCKETS

Classes

_ThreadStream

sys.stdout/sys.stderr that sends each thread's writes to its own buffer.

Agent

Serve plesty.bench clients for one bench.

Functions

_thread_stream(→ _ThreadStream)

The process's sys.<name> as a _ThreadStream, installed on first use.

_captured(→ Any)

Capture what this thread prints, and only this thread; yields (out, err).

serve_in_thread(→ tuple[threading.Thread, threading.Event])

Run agent on a daemon thread; returns the thread and its stop event.

_ok(→ dict[str, Any])

_error(→ dict[str, Any])

_frame(→ bytes)

_relay_socket(→ Any)

Return this thread's socket to address, connecting on first use.

_relay(→ dict[str, Any])

Send one device-server request to address and return its reply envelope.

_drop_relay_socket(→ None)

Close and forget this thread's socket to address.

_version(→ str)

Module Contents

plesty.server.services.agent.DEFAULT_BIND = 'tcp://*:5550'
plesty.server.services.agent.TOKEN_ENV = 'PLESTY_BENCH_TOKEN'
plesty.server.services.agent.FORBIDDEN
plesty.server.services.agent.FETCH_LIMIT = 16777216
plesty.server.services.agent.DEVICE_TIMEOUT_MS = 15000
class plesty.server.services.agent._ThreadStream(original: Any)

Bases: io.TextIOBase

sys.stdout/sys.stderr that sends each thread’s writes to its own buffer.

exec captures what a command prints, and requests run on several worker threads at once. contextlib.redirect_stdout swaps the stream for the whole process, so two commands running together wrote into each other’s capture: a client’s jobs cancel was answered with the [] of a console’s jobs list --json poll, and an install that was being followed got an empty jobs log --json and died on it (#19). This stands in for the stream once, and a thread that is capturing gets its own buffer; every other thread writes through to the original stream.

Initialize self. See help(type(self)) for accurate signature.

Parameters:

original (Any)

original
local
writable() → bool

Return whether object was opened for writing.

If False, write() will raise OSError.

Return type:

bool

_target() → Any
Return type:

Any

write(text: str) → int

Write string s to stream.

Return the number of characters written (which is always equal to the length of the string).

Parameters:

text (str)

Return type:

int

flush() → None

Flush write buffers, if applicable.

This is not implemented for read-only and non-blocking streams.

Return type:

None

isatty() → bool

Return whether this is an ‘interactive’ stream.

Return False if it can’t be determined.

Return type:

bool

property encoding: str
Return type:

str

plesty.server.services.agent._STREAMS_LOCK
plesty.server.services.agent._thread_stream(name: str) → _ThreadStream

The process’s sys.<name> as a _ThreadStream, installed on first use.

Parameters:

name (str)

Return type:

_ThreadStream

plesty.server.services.agent._captured() → Any

Capture what this thread prints, and only this thread; yields (out, err).

Return type:

Any

plesty.server.services.agent._SOCKETS
class plesty.server.services.agent.Agent(home: plesty.server.model.home.Home, fleet_path: str | os.PathLike[str] | None = None, bind: str = DEFAULT_BIND, token: str | None = None, workers: int = 4)

Serve plesty.bench clients for one bench.

Configure the agent.

Parameters:

  • home (plesty.server.model.home.Home) – The bench home every command runs against.

  • fleet_path (str | os.PathLike[str] | None) – Fleet file passed to every command (-f), or None.

  • bind (str) – ZMQ endpoint to bind.

  • token (str | None) – Shared secret; None reads PLESTY_BENCH_TOKEN; empty means no check (a bench on a trusted network).

  • workers (int) – Concurrent requests.

home
fleet_path
bind = 'tcp://*:5550'
token
workers = 4
bound: str | None = None
served = 0
device_timeout_ms = 15000
handle(message: Any) → dict[str, Any]

Answer one decoded request (pure; no socket involved).

Parameters:

message (Any)

Return type:

dict[str, Any]

description() → dict[str, Any]

What describe returns.

Return type:

dict[str, Any]

exec(argv: collections.abc.Sequence[Any]) → dict[str, Any]

Run plesty-server <argv> in-process against this bench; capture everything.

Parameters:

argv (collections.abc.Sequence[Any])

Return type:

dict[str, Any]

device(name: str, request: Any) → dict[str, Any]

Relay one device-server request to the server of name, and its reply back.

Why a request type of its own, when the rule is “the CLI is the API”: a control panel is a long-lived session of many small round trips — a poll per parameter per second — and exec would spawn the click application for each one. This carries the device wire itself, so the panel a console opens locally is the same panel driven from a laptop.

Why relayed at all, when device servers bind tcp://* and could be reached directly: they check no token. The agent does, so the bench keeps one authenticated door instead of one per instrument.

The reply is passed through as the device server framed it, so an error from the instrument reaches the caller as its own error, not as a relay failure.

Parameters:

  • name (str) – The declared device whose server to reach.

  • request (Any) – The device-server request (query, write, call, describe, help) to forward verbatim.

Returns:

The reply envelope to send back to the client.

Raises:

ValueError – When the name or the request is missing or malformed.

Return type:

dict[str, Any]

fetch(path: str) → dict[str, Any]

Read a file under the bench home; path is relative to it.

Parameters:

path (str)

Return type:

dict[str, Any]

serve(stop: threading.Event, poll_ms: int = 100, max_replies: int | None = None) → int

Bind and answer requests until stop is set; returns the number answered.

Parameters:

  • stop (threading.Event) – Set to end the loop.

  • poll_ms (int) – Socket poll interval.

  • max_replies (int | None) – Stop once this many replies went out (tests).

Return type:

int

plesty.server.services.agent.serve_in_thread(agent: Agent, stop: threading.Event | None = None) → tuple[threading.Thread, threading.Event]

Run agent on a daemon thread; returns the thread and its stop event.

Parameters:

  • agent (Agent)

  • stop (threading.Event | None)

Return type:

tuple[threading.Thread, threading.Event]

plesty.server.services.agent._ok(result: Any) → dict[str, Any]
Parameters:

result (Any)

Return type:

dict[str, Any]

plesty.server.services.agent._error(message: str, kind: str) → dict[str, Any]
Parameters:

  • message (str)

  • kind (str)

Return type:

dict[str, Any]

plesty.server.services.agent._frame(reply: dict[str, Any]) → bytes
Parameters:

reply (dict[str, Any])

Return type:

bytes

plesty.server.services.agent._relay_socket(address: str, timeout_ms: int) → Any

Return this thread’s socket to address, connecting on first use.

Parameters:

  • address (str) – The device server’s endpoint.

  • timeout_ms (int) – Receive timeout.

Returns:

A connected ZMQ DEALER belonging to the calling thread.

Return type:

Any

plesty.server.services.agent._relay(address: str, request: dict[str, Any], timeout_ms: int) → dict[str, Any]

Send one device-server request to address and return its reply envelope.

The device server’s own {"status": ...} envelope is returned unchanged, so a device that refused a write says so in its own words. Only the relay’s own failures — a server that stopped answering, a reply that was not JSON — are reported as the agent’s.

Parameters:

  • address (str) – The device server’s endpoint.

  • request (dict[str, Any]) – The request to forward.

  • timeout_ms (int) – How long to wait for the reply.

Returns:

The reply envelope.

Return type:

dict[str, Any]

plesty.server.services.agent._drop_relay_socket(address: str) → None

Close and forget this thread’s socket to address.

Parameters:

address (str)

Return type:

None

plesty.server.services.agent._version() → str
Return type:

str