Python API

ResolvedIdentity

class ResolvedIdentity

An identity successfully resolved from an SSH public key.

Attributes: subject: Opaque, consumer-defined identifier — e.g. "player:42", "sre:alice". The gateway never parses this; it only forwards. claims: Free-form key/value map of additional attributes the consumer wants to send along (groups, display name, theme, etc.). fingerprint: The OpenSSH-style SHA256 fingerprint ("SHA256:…") that was resolved. Populated by the gateway; resolvers may leave it empty.

SSHKeyResolver

class SSHKeyResolver

Asynchronous map from SSH public key → application identity.

Implementations are called once per inbound SSH connection, during the validate_public_key phase of the handshake. Return None to signal “this key is not known to me” — the gateway then either falls through to password auth or rejects the connection depending on the require_resolver gateway setting.

Implementations must be coroutine-safe. A sync implementation can be exposed async-style by wrapping its work in :func:asyncio.to_thread.

NullResolver

class NullResolver

Resolver that never resolves anything.

Equivalent to not configuring a resolver at all — exists so callers can pass a non-None resolver unconditionally. resolve always returns None, so the gateway falls through to password auth (or rejects if require_resolver is True — typically a misconfiguration).

AuthorizedKeysFileResolver

class AuthorizedKeysFileResolver

Resolve identities against a file in OpenSSH authorized_keys format.

Each line in the file is a standard OpenSSH public-key entry::

[options] <keytype> <base64-payload> [comment]

Optional options are comma-separated key="value" / key flags (OpenSSH’s sshd_config(5) grammar). Recognised here are:

• subject="…" — explicit subject to use; defaults to the comment if absent, or "key:<fp>" if neither is set.
• claim-<name>="…" — a single claim entry on the resulting identity. Multiple allowed; e.g. claim-role="oncall",claim-display="alice".

Unrecognised OpenSSH options (no-pty, command="…", etc.) are preserved in the claims map under a _options key so the upstream consumer can inspect them if desired.

Example line::

subject="sre:alice",claim-role="oncall" ssh-ed25519 AAAAC3N… alice@laptop

The file is read and parsed lazily on each resolve call — that keeps this class simple and makes key-rotation pick up immediately. For deployments with a huge authorized_keys file, wrap this in a caching resolver.

_AuthorizedKeyEntry

class _AuthorizedKeyEntry

TransportSession

class TransportSession

Transport + pyte terminal emulation behind the Session protocol.

Satisfies the :class:~provide.uterm.io.Session protocol: snapshot(), send(), wait_for_update().

Args: transport: A connected-or-connectable :class:~provide.uterm.transports.base.ConnectionTransport. cols: Terminal width (default 80). rows: Terminal height (default 25). send_encoding: Codec used by :meth:send to encode outgoing strings ("utf-8" by default; telnet uses "cp437"). Encoding always uses errors="replace" so unrepresentable characters never raise.

snapshot

def snapshot(...)

Return the current emulated screen state.

ansi_screen

def ansi_screen(...)

Return the current screen as ANSI-styled text (with SGR colors).

Delegates to :meth:TerminalEmulator.ansi_screen. Use this when shipping a snapshot to a live renderer (xterm.js dashboard, AnsiBuffer spy) so colors survive — :meth:snapshot returns plain text only.

is_connected

def is_connected(...)

Return True if the session is connected.

screen_change_seq

def screen_change_seq(...)

Return a monotonic counter that increments on each screen update.

Capture this before sending input, then pass the value to :meth:wait_for_screen_change to avoid reading stale screen data.

add_watch

def add_watch(...)

Register a callback fired with each raw byte chunk read from the wire.

This is the supported tap for raw terminal bytes on every transport session (telnet and websocket): register the watcher here to receive IAC-stripped, ANSI/CP437-intact bytes before pyte processes them. For examples, see :meth:add_watch and session.add_watch(lambda state, raw: buf.extend(raw)).

Called from _reader_loop immediately after IAC stripping and before the emulator processes the bytes — so the chunk still contains every ANSI SGR escape, cursor-positioning sequence and CP437 high byte that arrived from the server. Useful for fanning terminal output (with colors intact) to a hijack hub or recording tee, since :meth:snapshot returns pyte’s plain-text decoded display which has already absorbed the escape sequences.

Args: callback: (state_dict, raw_bytes) -> None. state_dict is currently always empty; the second positional carries the byte chunk. Callbacks must NOT block — schedule any async work onto a queue / task. interval_s: Reserved for future throttled-fan-out modes; currently ignored.

add_control_frame_watch

def add_control_frame_watch(...)

Register a callback fired with each inline control frame’s payload.

Only invoked when the session was constructed with control_frames=True; a no-op registration otherwise (there is nothing to dispatch since the decoder is never engaged). The payload is the parsed JSON object (e.g. {"type": "render_speed", "cps": 2400}) — the raw framing bytes never reach the emulator or :meth:add_watch callbacks in this mode.

Args: callback: (control_payload) -> None. Must not block.

ExpectSession

class ExpectSession

snapshot

def snapshot(...)

screen_change_seq

def screen_change_seq(...)

ExpectResult

class ExpectResult

Result from :func:send_and_expect.

LineEditor

class LineEditor

Generic line editor for terminal sessions.

Accumulates input characters until Enter is pressed, with support for readline-style editing shortcuts and password masking.

Features: - Character-by-character buffering until Enter/Return - Full cursor position tracking enabling mid-line editing - Backspace/Delete handling (removes character before cursor) - Readline shortcuts: Ctrl+A (start of line), Ctrl+E (end of line), Ctrl+U (kill backward to start), Ctrl+K (kill forward to end), Ctrl+B (left one char), Ctrl+F (right one char), Ctrl+W (kill word backward) - Password masking: echoes ‘*’ instead of actual characters - Configurable maximum line length (prevents DoS) - Optional async write callback for terminal output

Terminal Assumptions: - Assumes VT100-compatible terminal (ANSI escape codes) - This is true for all BBS systems (Telnet, SSH, WebSocket) - Cursor movement uses relative ANSI sequences so the editor does not need to know the screen column where input began

Args: max_length: Maximum number of characters to accept (default 80). password_mode: If True, mask input with asterisks (default False). on_write: Async callback(data: str) for terminal output. Called for all output including echoes and cursor movements. Exceptions propagate to caller. If None, no output is sent (silent mode).

Example: »> async def on_write(data: str) -> None: … await session.send(data) »> editor = LineEditor(max_length=40, password_mode=False, … on_write=on_write) »> line = None »> for ch in user_input: … line = await editor.process_char(ch) … if line is not None: … print(f"Got line: {line}")

reset

def reset(...)

Reset the buffer and cursor to empty state.

get_buffer

def get_buffer(...)

Get current buffer contents.

set_max_length

def set_max_length(...)

Change the maximum line length.

set_password_mode

def set_password_mode(...)

Enable or disable password masking.

TerminalEmulator

class TerminalEmulator

VT/ANSI terminal emulator backed by pyte.

Args: cols: Terminal width in columns (default 80). rows: Terminal height in rows (default 25). term: Terminal type string (default "ANSI").

Memory / scrollback bounds: Backed by pyte.Screen, which is bounded: only the visible viewport (cols * rows cells) is retained. Scrolling overwrites rather than buffering. There is no off-screen scrollback in this layer — applications that need history must record the raw byte stream separately (see :mod:provide.uterm.replay). This means get_snapshot and ansi_screen always allocate O(colsrows), independent of session age. resize is also O(colsrows): pyte re-flows the buffer, clipping rows that no longer fit.

process

def process(...)

Feed raw bytes (CP437) through the emulator.

Args: data: Raw bytes from a transport or file.

get_raw_tail

def get_raw_tail(...)

Return the bounded rolling tail of raw decoded output (ANSI intact).

get_snapshot

def get_snapshot(...)

Return the current screen state.

Returns a dict with:

• screen: Full screen text (newline-separated rows).
• screen_hash: SHA-256 of the screen text.
• cursor: {"x": int, "y": int}.
• cols, rows, term.
• cursor_at_end: True if cursor is at or past the last content line.
• has_trailing_space: True if the screen ends with a space or colon.
• captured_at: Unix timestamp of this snapshot (always fresh).

ansi_screen

def ansi_screen(...)

Return the current screen as a single string with ANSI SGR codes.

Walks pyte’s per-cell style buffer and emits SGR escape sequences whenever the style changes between adjacent cells. Use this when a downstream consumer (xterm.js dashboard, AnsiBuffer in a spy) needs the visual state including colors — :meth:get_snapshot’s plain screen field discards pyte’s style attributes.

Rows are joined with \n; each row ends with \x1b[0m so a consumer’s subsequent writes start from a clean attribute state.

reset

def reset(...)

Reset terminal to its initial state.

resize

def resize(...)

Resize the terminal.

Args: cols: New width in columns. rows: New height in rows.

RecordingStore

class RecordingStore

Protocol for persisting and retrieving session recordings.

Implement this protocol to provide a custom recording backend (e.g. S3, GCS, a database, or any remote object store). The lifecycle is:

1. start_session – called once when a session begins recording. Persist metadata keyed by session_id.
2. append_events – called repeatedly with batches of JSON-serialisable event dicts. Each dict has at minimum ts, event, and data.
3. end_session – called once when the session stops recording.

Query methods (recording_meta, get_entries, get_path) may be called at any time, including while the session is still active.

See InMemoryRecordingStore for a minimal reference implementation.

LocalFileRecordingStore

class LocalFileRecordingStore

File-backed implementation of RecordingStore using JSONL files.

InMemoryRecordingStore

class InMemoryRecordingStore

In-memory implementation of RecordingStore.

Keeps all events in Python lists. Useful for tests and as a reference implementation showing the expected behaviour for custom remote stores.

Note: data is lost when the process exits. Do not use in production unless you only need ephemeral recordings.

NullRecordingStore

class NullRecordingStore

No-op implementation of RecordingStore.

All writes are silently discarded and all reads return empty results. Use this when recording is disabled – it eliminates None checks throughout the calling code while keeping the RecordingStore interface consistent.

TerminalReader

class TerminalReader

Structural type for an async byte-stream reader.

TerminalWriter

class TerminalWriter

Structural type for an async byte-stream writer.

write

def write(...)

get_extra_info

def get_extra_info(...)

close

def close(...)

SessionLogger

class SessionLogger

Async session recorder using a pluggable RecordingStore.

Each log entry is a JSON object on its own line with at minimum: {"ts": ..., "event": ..., "data": {...}}.

set_context

def set_context(...)

Set metadata context for subsequent log entries.

clear_context

def clear_context(...)

Clear metadata context.

Session

class Session

Minimal interface expected by :class:PromptWaiter and :class:InputSender.

snapshot

def snapshot(...)

Return latest snapshot without performing network I/O.

PromptWaiter

class PromptWaiter

Wait for a prompt to appear in the session snapshot.

Args: session: BBS session object conforming to :class:Session. on_screen_update: Optional callback invoked with the raw screen text on each poll.

InputSender

class InputSender

Send keystrokes to a session respecting input type semantics.

Args: session: BBS session object conforming to :class:Session.

TelnetSession

class TelnetSession

Telnet transport with pyte terminal emulation.

Satisfies the :class:~provide.uterm.io.Session protocol: snapshot(), send(), wait_for_update().

Uses :class:TelnetTransport (not raw :class:TelnetClient) for full RFC 854 IAC negotiation — required by TWGS and other BBS servers.

Use :func:connect_telnet for a convenient factory, or construct directly and call :meth:connect.

ControlFrameProtocolError

class ControlFrameProtocolError

Raised when an inline control frame is malformed.

DataChunk

class DataChunk

Decoded terminal data from the inline stream.

kind

def kind(...)

ControlChunk

class ControlChunk

Decoded control payload from the inline stream.

kind

def kind(...)

ControlFrameDecoder

class ControlFrameDecoder

Incrementally decode the inline DLE/STX control-frame stream.

feed

def feed(...)

Decode all complete events from chunk and buffer the rest.

finish

def finish(...)

Decode any remaining buffered data and reject truncated control frames.

TerminalDefaults

class TerminalDefaults

Default host/port values used across provide-uterm transports and gateways.

All constants are class-level and can be referenced as TerminalDefaults.X without instantiation. Override at the call site rather than modifying this class directly.

token_file

def token_file(...)

Default resume-token file path (~/.uterm/session_token).

ChannelHello

class ChannelHello

Client-advertised typed-channel versions.

NegotiatedChannels

class NegotiatedChannels

Per-connection typed-channel grants and sequence counters.

granted

def granted(...)

Return a copy of the currently granted channels.

is_negotiated

def is_negotiated(...)

Return whether channel is negotiated, defaulting to the configured primary channel.

handle_hello

def handle_hello(...)

Negotiate channel versions and return a hello_ack payload.

next_seq

def next_seq(...)

Increment and return the sequence number for channel.

export_grants

def export_grants(...)

Return a serializable granted-channel map.

restore_grants

def restore_grants(...)

Restore persisted grants and reset sequence counters for a fresh channel instance.

LinkPattern

class LinkPattern

An immutable descriptor for one server-driven clickable text pattern.

Parameters

pattern: JavaScript regex source string (e.g. r"\((\d{1,5})\)"). action: What happens when the user clicks: "cmd", "url", "key", or "focus". id: Optional stable identifier used by :meth:LinkPatternRegistry.unregister. If two patterns share the same id the later register call silently replaces the earlier one. flags: Regex flags forwarded to new RegExp(pattern, flags). Defaults to "g"; "g" is always ensured on the client side regardless. group: Which capture group is the clickable span (0 = whole match). payload: Click payload template; $1, $2, … are substituted from captures. hover: Hover tooltip template (same $N substitution). class_: CSS class name applied to highlighted ranges. Serialised as "class" in the wire frame (class is a Python reserved word).

to_frame_entry

def to_frame_entry(...)

Serialise to the wire-format dict expected by xterm-server-links.js.

Only non-default / non-empty optional fields are included to keep frames compact. class_ is emitted as "class".

LinkPatternRegistry

class LinkPatternRegistry

Active pattern set for one owner (session, worker, etc.).

Patterns are stored in insertion order. Registering a pattern whose id already exists replaces the earlier pattern in-place (preserving the slot’s position so that order is predictable for callers who register once and later refresh the same id).

Patterns without an id (id=None) are appended and cannot be removed individually; use :meth:clear to reset the whole set.

register

def register(...)

Add pattern to the active set.

If pattern has an id and that id is already registered the existing entry is replaced (same dict slot, so insertion order is preserved for that id). If id is None the pattern is appended unconditionally.

unregister

def unregister(...)

Remove the pattern registered under pattern_id.

Returns True if the pattern was found and removed, False if no pattern with that id exists.

clear

def clear(...)

Remove all patterns and reset the id-less counter.

get_all

def get_all(...)

Return all active patterns in insertion order.

sync_payload

def sync_payload(...)

Return the ready-to-send dict for :func:~provide.uterm.control_channel.encode_control_frame.

The returned dict has the shape::

{"type": "link_patterns", "patterns": [...]}

Calling this method is non-destructive; the registry state is unchanged.

ByteReader

class ByteReader

Minimal async reader — only read(n) is required.

WebSocketSession

class WebSocketSession

WebSocket transport with pyte terminal emulation.

Satisfies the :class:~provide.uterm.io.Session protocol: snapshot(), send(), wait_for_update().

ScriptedTelnetUpstream

class ScriptedTelnetUpstream

Deterministic telnet wire for embed tests (no TCP).

is_connected

def is_connected(...)

sent_wire

def sent_wire(...)

InterceptContext

class InterceptContext

ByteInterceptor

class ByteInterceptor

PassThroughInterceptor

class PassThroughInterceptor

MemoryUpstream

class MemoryUpstream

Deterministic test upstream.

is_connected

def is_connected(...)

sent

def sent(...)

complete_remote

def complete_remote(...)

ClientHandle

class ClientHandle

_Deferred

class _Deferred

EmbedSession

class EmbedSession

Ordered multi-client proxy session (re-entrancy-safe via single lock).

on_application_data

def on_application_data(...)

on_client_data

def on_client_data(...)

on_wire

def on_wire(...)

on_lifecycle

def on_lifecycle(...)

SessionLifecycle

class SessionLifecycle

InterceptAction

class InterceptAction

BackpressurePolicy

class BackpressurePolicy

ByteDirection

class ByteDirection

WireEventKind

class WireEventKind

InterceptResult

class InterceptResult

pass_

def pass_(...)

replace

def replace(...)

consume

def consume(...)

defer

def defer(...)

inject

def inject(...)

ClientMetadata

class ClientMetadata

ClientFilter

class ClientFilter

matches

def matches(...)

UpstreamPipe

class UpstreamPipe

is_connected

def is_connected(...)

TelnetPolicy

class TelnetPolicy

terminal_type

def terminal_type(...)

window_size

def window_size(...)

on_option

def on_option(...)

on_subnegotiation

def on_subnegotiation(...)

DefaultTelnetPolicy

class DefaultTelnetPolicy

on_option

def on_option(...)

on_subnegotiation

def on_subnegotiation(...)

EmbedHub

class EmbedHub

In-process session factory.

session_ids

def session_ids(...)

get_session

def get_session(...)

remove_session

def remove_session(...)

HijackSession

class HijackSession

A live hijack lease.

AcquireResult

class AcquireResult

HijackCoordinator

class HijackCoordinator

Single-writer hijack arbitration for one worker session.

session

def session(...)

acquire

def acquire(...)

Acquire a hijack lease, always generating a new hijack_id.

If the same owner already holds the lease the lease is renewed with a fresh hijack_id so callers always receive an authoritative token for the new lease period. A different owner while a lease is active returns ok=False.

heartbeat

def heartbeat(...)

release

def release(...)

can_send_input

def can_send_input(...)

_FrameBase

class _FrameBase

Base for all frame models — forbid unknown fields so producers and consumers can’t silently drift.

TermFrame

class TermFrame

Raw terminal output bytes from the worker to subscribers.

InputFrame

class InputFrame

Browser/operator input destined for the worker.

SnapshotReqFrame

class SnapshotReqFrame

Browser-originated request for a fresh screen snapshot.

SnapshotFrame

class SnapshotFrame

Worker-originated full-screen snapshot.

ControlFrame

class ControlFrame

Server-originated worker-control frame (pause/resume/step).

HijackStateFrame

class HijackStateFrame

Broadcast lease-state update.

HijackRequestFrame

class HijackRequestFrame

Browser-originated request to acquire the hijack lease.

HijackReleaseFrame

class HijackReleaseFrame

Browser-originated request to release the hijack lease.

HijackStepFrame

class HijackStepFrame

Browser-originated single-step request.

WorkerConnectedFrame

class WorkerConnectedFrame

WorkerDisconnectedFrame

class WorkerDisconnectedFrame

WorkerHelloFrame

class WorkerHelloFrame

Worker-originated hello-frame carrying input_mode + capabilities.

HeartbeatFrame

class HeartbeatFrame

HeartbeatAckFrame

class HeartbeatAckFrame

Server reply to a browser heartbeat — refreshes the lease.

PingFrame

class PingFrame

PongFrame

class PongFrame

HelloFrame

class HelloFrame

Server-originated hello-frame to the browser describing capabilities.

Schema is intentionally permissive (extra="ignore") because the field set drifts as new capabilities land; field-by-field tightening will happen once the wire format is fully stable.

ResumeFrame

class ResumeFrame

IdentityFrame

class IdentityFrame

Inline control-channel identity frame.

SessionTokenFrame

class SessionTokenFrame

ResumeOkFrame

class ResumeOkFrame

ResumeFailedFrame

class ResumeFailedFrame

LinkPatternEntry

class LinkPatternEntry

LinkPatternsFrame

class LinkPatternsFrame

AnalysisFrame

class AnalysisFrame

ErrorFrame

class ErrorFrame

StatusFrame

class StatusFrame

Worker-originated status passthrough (coerce_worker_status_frame).

Schema is permissive because the worker may attach arbitrary status payloads. The frame type discriminator and ts field are the only guarantees.

InputModeChangedFrame

class InputModeChangedFrame

ApprovalPendingFrame

class ApprovalPendingFrame

ApprovalResolvedFrame

class ApprovalResolvedFrame

PresenceUpdateFrame

class PresenceUpdateFrame

DeckMux per-user presence update — schema is permissive because optional fields (scroll, selection, pin, typing, queued_keys) are attached only when relevant.

PresenceSyncFrame

class PresenceSyncFrame

Full presence roster sent on browser connect.

PresenceLeaveFrame

class PresenceLeaveFrame

ControlTransferFrame

class ControlTransferFrame

DeckMux ownership-transfer notice.

HijackAcquireResponse

class HijackAcquireResponse

HijackHeartbeatResponse

class HijackHeartbeatResponse

HijackStepResponse

class HijackStepResponse

HijackReleaseResponse

class HijackReleaseResponse

HijackSnapshotResponse

class HijackSnapshotResponse

HijackSendResponse

class HijackSendResponse

HijackEventsResponse

class HijackEventsResponse

SessionStatusResponse

class SessionStatusResponse

Shape of GET /api/sessions/{id} and items in GET /api/sessions.

Mirrors provide-uterm-server SessionRuntimeStatus.

SessionSnapshotResponse

class SessionSnapshotResponse

Shape of GET /api/sessions/{id}/snapshot response.

SessionEventsResponse

class SessionEventsResponse

Shape of GET /api/sessions/{id}/events response.

SessionModeResponse

class SessionModeResponse

Shape of POST /api/sessions/{id}/mode response.

SessionAnalyzeResponse

class SessionAnalyzeResponse

Shape of POST /api/sessions/{id}/analyze response.

RecordingMetaResponse

class RecordingMetaResponse

Shape of GET /api/sessions/{id}/recording response.

RecordingEntry

class RecordingEntry

Single entry from GET /api/sessions/{id}/recording/entries.

Frame

class Frame

Canonical shape of a control-channel or data-channel frame.

HijackableMixin

class HijackableMixin

Mixin that makes an async worker class hijackable by a human operator.

Adds pause/resume/step/watchdog primitives. Intended usage — add as a base class to your worker class, then call :meth:await_if_hijacked at checkpoints in your automation loop::

class MyBot(HijackableMixin):
    async def run_loop(self) -> None:
        while True:
            await self.await_if_hijacked()  # pauses here when hijacked
            await self.do_action()

The hub or manager calls :meth:set_hijacked to pause/resume. The dashboard calls :meth:request_step to allow one loop iteration while paused.

note_progress

def note_progress(...)

Signal that the worker is making progress (resets the watchdog timer).

Call this whenever meaningful work occurs (e.g. after each turn, screen change, or successful action) to prevent the watchdog from firing.

start_watchdog

def start_watchdog(...)

Start a background task that triggers on_stuck if the worker stops progressing.

The watchdog fires when no call to :meth:note_progress has been seen for stuck_timeout_s seconds. While hijacked, the timer is suppressed.

Args: stuck_timeout_s: Seconds without progress before firing. check_interval_s: How often to check (default 5 s). on_stuck: Async callback called when stuck. Typical use: disconnect the session so the outer reconnect loop triggers.

Segment

class Segment

A run of text sharing one foreground color + bold flag.

AnsiBuffer

class AnsiBuffer

Virtual terminal backed by a pyte Screen.

Feed raw bytes with :meth:feed and retrieve ANSI-styled output lines with :meth:render_lines.

resize

def resize(...)

reset

def reset(...)

feed

def feed(...)

render_lines

def render_lines(...)

PromptDetector

class PromptDetector

Intelligent prompt detection with cursor-awareness.

pattern_count

def pattern_count(...)

Return the number of compiled patterns.

compile_failures

def compile_failures(...)

Return immutable view of pattern compile failures.

Each entry is {"id": str, "regex": str | None, "error": str}. In strict=True mode this tuple is always empty (the constructor would have raised before returning).

prompt_region

def prompt_region(...)

Extract a bottom-of-content region likely to contain prompts.

Returns (region_text, cursor_in_region).

We anchor to the last non-empty line of the screen, not the bottom row, because many UIs leave blank rows below the last content.

normalize_prompt_region

def normalize_prompt_region(...)

Normalize volatile prompt-region fields for stable fingerprinting.

prompt_fingerprint

def prompt_fingerprint(...)

Compute a stable fingerprint for prompt-detection caching.

detect_prompt

def detect_prompt(...)

Detect if snapshot contains a prompt waiting for input.

This method keeps the legacy API and returns only the match. Use detect_prompt_with_diagnostics() to also get partial-match reasons.

Args: snapshot: Screen snapshot with timing and cursor metadata

Returns: PromptMatch if a prompt pattern matches, None otherwise

detect_prompt_with_diagnostics

def detect_prompt_with_diagnostics(...)

Detect prompt and include partial-match diagnostics.

Args: snapshot: Screen snapshot with timing and cursor metadata

Returns: PromptDetectionDiagnostics containing both match and partial-match failures

add_pattern

def add_pattern(...)

Add a new pattern to the detector.

Args: pattern: Pattern dictionary to add

reload_patterns

def reload_patterns(...)

Replace all patterns with new set.

Args: patterns: New list of pattern dictionaries

PatternDetector

class PatternDetector

DetectorPatternCompileError

class DetectorPatternCompileError

Raised in strict=True mode when a pattern fails to compile.

The non-strict default merely logs failures and continues with the surviving patterns — useful for “soft” environments where a broken rule shouldn’t take the whole detector offline. Production deploys that load curated rules should pass strict=True so a typo in a rules file is caught at startup instead of silently degrading detection.

ScreenSnapshot

class ScreenSnapshot

Contract for the snapshot dict passed to process_screen.

PromptMatch

class PromptMatch

A matched prompt pattern with its rule metadata.

PromptDetection

class PromptDetection

Complete prompt detection result.

PromptDetectionDiagnostics

class PromptDetectionDiagnostics

Detection result with partial-match diagnostics for debugging.

FlowStep

class FlowStep

Decision returned by :meth:FlowEngine.advance.

FlowEngine

class FlowEngine

Advance named flows using existing prompt detectors and rule metadata.

advance

def advance(...)

Return the next action for flow_id on the current screen.

When several flow steps’ gate prompts match (e.g. a stale prompt left in scrollback above the live one), the prompt whose match sits closest to the tail — the current cursor region — wins, so scrollback does not beat the live prompt. Ties keep the earliest flow step.

RegexRule

class RegexRule

to_regex

def to_regex(...)

ScreenConstraint

class ScreenConstraint

KVExtractRule

class KVExtractRule

PromptRule

class PromptRule

MenuOption

class MenuOption

MenuRule

class MenuRule

TimingRule

class TimingRule

ActionRule

class ActionRule

FlowRule

class FlowRule

RuleSet

class RuleSet

to_prompt_patterns

def to_prompt_patterns(...)

from_json_file

def from_json_file(...)

RuleLoadResult

class RuleLoadResult

KVExtractor

class KVExtractor

Extract structured key-value data from screen text.

extract

def extract(...)

Extract key-value data from screen using configured patterns.

Args: screen: Screen text to extract from kv_config: Extraction configuration from prompt pattern Can be a single field config or list of field configs run_validation: Whether to validate extracted values (default True)

Returns: Dictionary of extracted values, None if config invalid or extraction failed May include “_validation” key with validation results

DetectionEngine

class DetectionEngine

Rule-based prompt detection and data extraction engine.

Accepts rules at init, compiles patterns, and provides both sync _sync_process_screen() and async process_screen() for prompt detection

• KV extraction. The async variant also handles buffering, idle detection, screen saving, and callable hooks.

add_hook

def add_hook(...)

Register an async hook called after each process_screen() call.

Hook signature: async def hook(snapshot, detection, buffer, is_idle)

is_idle

def is_idle(...)

True if the screen has been stable for >= idle_threshold_s.

namespace

def namespace(...)

Game/namespace identifier.

set_namespace

def set_namespace(...)

Update namespace and propagate to the ScreenSaver if present.

get_screen_saver_status

def get_screen_saver_status(...)

Return screen-saver status dict.

set_screen_saving

def set_screen_saving(...)

Enable or disable the ScreenSaver.

debug_state

def debug_state(...)

Return internal debug info (avoids direct private-attr access by callers).

detect_with_diagnostics

def detect_with_diagnostics(...)

Detect with partial-match info for debugging.

reload_rules

def reload_rules(...)

Hot-reload rules. Transactional: on failure, old rules remain active.

Raises: ValueError: If new rules cannot be loaded.

detector

def detector(...)

Access the underlying PromptDetector.

pattern_count

def pattern_count(...)

Number of compiled patterns.

enabled

def enabled(...)

Whether the engine processes screens.

enabled

def enabled(...)

ScreenSaver

class ScreenSaver

Saves unique screens to disk in organized directory structure.

set_enabled

def set_enabled(...)

Enable or disable screen saving.

set_namespace

def set_namespace(...)

Set namespace for screen organization.

get_screens_dir

def get_screens_dir(...)

Get directory for saving screens.

Returns: Path to screens directory

save_screen

def save_screen(...)

Save screen snapshot to disk.

Blocking I/O warning: this method performs synchronous disk writes and directory creation. It is called from DetectionEngine.process_screen() (an async def), which means it blocks the event loop on every save. At low save rates (a few per second) this is acceptable; at high rates consider offloading via asyncio.get_event_loop().run_in_executor(None, ...).

Args: snapshot: Screen snapshot with screen, screen_hash, captured_at, etc. prompt_id: Optional prompt ID if detected force: Force save even if hash already saved

Returns: Path to saved screen file, or None if not saved

clear_saved_hashes

def clear_saved_hashes(...)

Clear the set of saved screen hashes.

Useful for forcing re-save of all screens.

get_saved_count

def get_saved_count(...)

Get count of saved unique screens.

Returns: Number of unique screen hashes saved

ScreenBuffer

class ScreenBuffer

Represents a buffered screen snapshot with timing metadata.

BufferManager

class BufferManager

Manages screen history buffer with timing calculation.

add_screen

def add_screen(...)

Add screen snapshot to buffer and calculate timing metadata.

Args: snapshot: Screen snapshot from terminal emulator

Returns: ScreenBuffer with timing metadata

get_recent

def get_recent(...)

Get N most recent buffered screens.

Args: n: Number of recent screens to retrieve

Returns: List of most recent ScreenBuffer objects (oldest first)

detect_idle_state

def detect_idle_state(...)

Detect if screen has been stable (idle) for threshold period.

Args: threshold_seconds: Minimum seconds of stability to consider idle

Returns: True if screen has been unchanged for >= threshold

clear

def clear(...)

Clear the buffer and reset state.

CommandDispatcher

class CommandDispatcher

Parse and dispatch ushell command lines.

Args: ctx: Runtime context dict. Expected optional keys:

         ``list_kv_sessions``
             Async callable ``() -> list[dict]`` — KV session list.
         ``env``
             CF env object with KV/DO bindings.
         ``storage``
             DO storage object (ctx.storage).

sandbox: :class:`~provide.uterm.shell._sandbox.Sandbox` instance
         for ``py`` commands.  A fresh one is created if omitted.

AnimatedResult

class AnimatedResult

Return type for animated render output — caller handles frame timing.

BrowserControlMessage

class BrowserControlMessage

WorkerControlMessage

class WorkerControlMessage

ControlPlane

class ControlPlane

ControlPlaneRef

class ControlPlaneRef

Stable identifier for a control-plane backend.

ControlPlaneConfig

class ControlPlaneConfig

Bootstrap configuration for control-plane backends.

EngineCapabilities

class EngineCapabilities

Engine feature flags discovered at bootstrap time.

ControlPlaneError

class ControlPlaneError

Base error for control-plane bootstrap and transaction failures.

ControlPlaneConfigurationError

class ControlPlaneConfigurationError

Raised when control-plane configuration is invalid or incomplete.

ControlPlaneCapabilityError

class ControlPlaneCapabilityError

Raised when a caller requests a capability the engine does not expose.

ControlPlaneConflictError

class ControlPlaneConflictError

Raised on commit when a write conflicts with a concurrently committed transaction.

Mirrors the serialization failure the SQLite backend produces via BEGIN IMMEDIATE + a held transaction lock: two overlapping transactions that write the same key cannot both succeed. The memory backend detects this optimistically at commit time so that, e.g., a lease-acquire race yields exactly one winner on both backends.

TokenStore

class TokenStore

SessionTokenRecord

class SessionTokenRecord

ResumeTokenRecord

class ResumeTokenRecord

Transaction

class Transaction

MemoryApprovalStore

class MemoryApprovalStore

MemoryTokenStore

class MemoryTokenStore

MemoryState

class MemoryState

MemoryTransaction

class MemoryTransaction

MemoryLeaseStore

class MemoryLeaseStore

MemoryControlPlane

class MemoryControlPlane

In-memory control-plane backend with shared mutable state.

session_store

def session_store(...)

token_store

def token_store(...)

approval_store

def approval_store(...)

lease_store

def lease_store(...)

MemorySessionStore

class MemorySessionStore

SqliteApprovalStore

class SqliteApprovalStore

SqliteTokenStore

class SqliteTokenStore

SqliteTransaction

class SqliteTransaction

SqliteLeaseStore

class SqliteLeaseStore

SqliteConnectionError

class SqliteConnectionError

Raised when a SQLite control-plane connection cannot be initialized.

SqliteControlPlane

class SqliteControlPlane

Inert SQLite-backed control-plane shell.

session_store

def session_store(...)

token_store

def token_store(...)

approval_store

def approval_store(...)

lease_store

def lease_store(...)

SqliteSessionStore

class SqliteSessionStore

SqliteMigrationError

class SqliteMigrationError

Raised when the SQLite control-plane schema cannot be migrated.

LeaseStore

class LeaseStore

LeaseRecord

class LeaseRecord

ApprovalStore

class ApprovalStore

ApprovalRecord

class ApprovalRecord

SessionStore

class SessionStore

SessionRecord

class SessionRecord