TelTest Phase 1: Core Client & Protocol — Implementation Plan¶

Design Document: docs/designs/teltest/README.md Priority: P1 — Critical for regression safety Estimated Duration: 3 weeks across 6 sections (1.1–1.6)

Summary¶

This plan implements the foundational layer of the TelTest E2E testing framework: the standalone async telnet client and its supporting infrastructure. All code lives in packages/teltest/ with zero MAID dependencies — pure Python 3.12+ and asyncio only.

The implementation adds:

TelnetProtocol — Byte-level IAC handling, negotiation state machine, GMCP subnegotiation extraction
ANSIStripper — Removal of ANSI/VT100 escape sequences from output text
OutputBuffer — Line splitting, prompt detection (multi-signal: GA, regex, quiescence), ring buffer with read cursor tracking
MUDClient — The top-level async client tying together connection lifecycle, background reader task, send/receive, and buffer access

Existing infrastructure leveraged: None — this is a greenfield standalone package.

Package layout after this phase:

packages/teltest/
├── pyproject.toml
├── src/teltest/
│   ├── __init__.py
│   ├── protocol.py          # TelnetProtocol
│   ├── ansi.py              # ANSIStripper
│   ├── buffer.py            # OutputBuffer
│   ├── client.py            # MUDClient
│   ├── exceptions.py        # ConnectionClosed, etc.
│   └── types.py             # Shared types, constants, dataclasses
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_protocol.py
│   ├── test_ansi.py
│   ├── test_buffer.py
│   ├── test_client.py
│   └── test_integration.py
└── README.md

Section 1.1: Package Scaffolding & Shared Types¶

Package: teltest | Priority: P0 | Dependencies: none

1.1.1 Package Structure¶

[ ] Create directory tree: packages/teltest/src/teltest/ and packages/teltest/tests/
[ ] Create packages/teltest/pyproject.toml:
name = "teltest", version = "0.1.0"
requires-python = ">=3.12"
Zero runtime dependencies (pure stdlib + asyncio)
Dev dependencies: pytest, pytest-asyncio
Package source: src/teltest
[ ] Create packages/teltest/src/teltest/__init__.py with public exports:
MUDClient, TelnetProtocol, ANSIStripper, OutputBuffer
Match, TranscriptEntry, PromptSignal
ConnectionClosed, ExpectTimeout, UnexpectedMatch, NotConnected
[ ] Create packages/teltest/tests/__init__.py (empty)
[ ] Create packages/teltest/tests/conftest.py with shared fixtures (see 1.5.1)
[ ] Create packages/teltest/README.md with basic package description

1.1.2 Telnet Constants¶

File: packages/teltest/src/teltest/types.py

[ ] Define telnet byte constants as module-level int values:

# Telnet command bytes
IAC: int = 255    # Interpret As Command
DONT: int = 254
DO: int = 253
WONT: int = 252
WILL: int = 251
SB: int = 250     # Subnegotiation Begin
SE: int = 240     # Subnegotiation End
GA: int = 249     # Go Ahead
NOP: int = 241    # No Operation

# Telnet option codes
OPT_ECHO: int = 1
OPT_SGA: int = 3       # Suppress Go-Ahead
OPT_TTYPE: int = 24    # Terminal Type
OPT_NAWS: int = 31     # Negotiate About Window Size
OPT_MSDP: int = 69     # MUD Server Data Protocol
OPT_MSSP: int = 70     # MUD Server Status Protocol
OPT_MCCP2: int = 86    # MUD Client Compression Protocol v2
OPT_MXP: int = 91      # MUD eXtension Protocol
OPT_GMCP: int = 201    # Generic MUD Communication Protocol

# TTYPE subnegotiation
TTYPE_IS: int = 0
TTYPE_SEND: int = 1

1.1.3 Shared Dataclasses & Exceptions¶

File: packages/teltest/src/teltest/types.py (continued)

[ ] Define PromptSignal enum:

class PromptSignal(enum.Enum):
    GA = "ga"                # Telnet Go-Ahead received
    REGEX = "regex"          # Prompt regex matched
    QUIESCENCE = "quiescence"  # No new data for settle time

[ ] Define TranscriptEntry frozen dataclass:

@dataclasses.dataclass(frozen=True)
class TranscriptEntry:
    timestamp: float
    direction: Literal["send", "recv", "gmcp_send", "gmcp_recv"]
    text: str

File: packages/teltest/src/teltest/exceptions.py

[ ] Define ConnectionClosed(Exception):
Attributes: buffer: str — text received before connection closed
__init__(self, buffer: str) -> None
__str__ includes truncated buffer (last 500 chars)
[ ] Define ExpectTimeout(AssertionError):
Attributes: pattern: str, timeout: float, buffer: str
__init__(self, pattern: str, timeout: float, buffer: str) -> None
__str__ includes pattern, timeout, and formatted buffer dump (last 20 lines)
[ ] Define UnexpectedMatch(AssertionError):
Attributes: pattern: str, matched_text: str, buffer: str
__init__(self, pattern: str, matched_text: str, buffer: str) -> None
[ ] Define NotConnected(Exception):
Raised when send() or expect() is called on a disconnected client
__init__(self, message: str = "Client is not connected") -> None

Section 1.2: TelnetProtocol — IAC Handling & Negotiation State Machine¶

Package: teltest | Priority: P0 | Dependencies: 1.1

File: packages/teltest/src/teltest/protocol.py

The TelnetProtocol class is a stateful byte-level parser that processes raw TCP data and separates telnet control sequences from application text. It operates synchronously on byte buffers — no I/O, no async. The caller feeds it raw bytes and receives clean text plus any GMCP messages.

1.2.1 Negotiation State Machine¶

The parser uses a state machine to track its position within IAC sequences:

class _TelnetState(enum.Enum):
    DATA = "data"           # Normal text data
    IAC = "iac"             # Received IAC byte, waiting for command
    WILL = "will"           # Received IAC WILL, waiting for option
    WONT = "wont"           # Received IAC WONT, waiting for option
    DO = "do"               # Received IAC DO, waiting for option
    DONT = "dont"           # Received IAC DONT, waiting for option
    SB = "sb"               # Inside subnegotiation, collecting bytes
    SB_IAC = "sb_iac"       # Inside SB, received IAC (checking for SE)

1.2.2 TelnetProtocol Class¶

[ ] class TelnetProtocol:
[ ] __init__(self, *, negotiate_gmcp: bool = False) -> None:
- self._state: _TelnetState = _TelnetState.DATA
- self._negotiate_gmcp: bool = negotiate_gmcp
- self._sb_buffer: bytearray = bytearray() — accumulates subnegotiation bytes
- self._sb_option: int = 0 — which option the current SB is for
- self._text_buffer: bytearray = bytearray() — accumulates clean text output
- self._response_buffer: bytearray = bytearray() — IAC responses to send back
- self._gmcp_messages: list[tuple[str, dict[str, Any]]] — extracted GMCP messages
- self._ga_received: bool = False — set True when GA is received, cleared on read
- self._option_state: dict[int, dict[str, bool]] — tracks negotiated option state per option; keys: "local" (we WILL), "remote" (they WILL)
[ ] feed(self, data: bytes) -> bytes:
- Feed raw TCP bytes through the state machine
- Returns clean text bytes (application data with all IAC sequences removed)
- Populates self._response_buffer with IAC negotiation responses
- Populates self._gmcp_messages with any extracted GMCP messages
- Sets self._ga_received = True when GA (IAC GA) is encountered
- State machine transitions per byte:
- DATA + IAC → IAC
- DATA + any other → append to _text_buffer
- IAC + IAC → escaped IAC (0xFF literal), append to _text_buffer, → DATA
- IAC + WILL → WILL
- IAC + WONT → WONT
- IAC + DO → DO
- IAC + DONT → DONT
- IAC + SB → SB, clear _sb_buffer
- IAC + GA → set _ga_received = True, → DATA
- IAC + NOP → ignore, → DATA
- IAC + any other → ignore unknown command, → DATA
- WILL + option → call _handle_will(option), → DATA
- WONT + option → call _handle_wont(option), → DATA
- DO + option → call _handle_do(option), → DATA
- DONT + option → call _handle_dont(option), → DATA
- SB + first byte → store as _sb_option, remain in SB
- SB + IAC → SB_IAC
- SB + any other → append to _sb_buffer
- SB_IAC + SE → call _handle_subnegotiation(_sb_option, _sb_buffer), → DATA
- SB_IAC + IAC → escaped IAC inside SB, append 0xFF to _sb_buffer, → SB
- SB_IAC + any other → append to _sb_buffer, → SB (robustness)
- At end: drain _text_buffer into return value, clear _text_buffer
[ ] get_response(self) -> bytes:
- Returns and clears _response_buffer
- Called by MUDClient after each feed() to get bytes to send back to server
[ ] get_gmcp_messages(self) -> list[tuple[str, dict[str, Any]]]:
- Returns and clears _gmcp_messages list
[ ] consume_ga(self) -> bool:
- Returns current _ga_received and resets it to False
- Used by OutputBuffer to detect prompt via GA signal
[ ] is_option_active(self, option: int, *, local: bool = False) -> bool:
- Check if a specific option has been successfully negotiated
- local=False checks remote side (they WILL), local=True checks our side (we WILL)

1.2.3 Negotiation Handlers¶

[ ] _handle_will(self, option: int) -> None:
Server offers to enable an option
Accepted options (respond DO):
- OPT_ECHO — password hiding
- OPT_SGA — suppress go-ahead (standard)
- OPT_GMCP — only if self._negotiate_gmcp is True
Refused options (respond DONT):
- OPT_MCCP2 — compression adds complexity; decline
- OPT_MSDP, OPT_MXP, OPT_MSSP — not needed
- Everything else — safe default: refuse unknown
Update _option_state[option]["remote"] = True for accepted
Append bytes([IAC, DO, option]) or bytes([IAC, DONT, option]) to _response_buffer
Guard: If option already negotiated (already in _option_state as active), do NOT re-send response (prevents negotiation loops)
[ ] _handle_wont(self, option: int) -> None:
Server disabling an option
Update _option_state[option]["remote"] = False
Respond bytes([IAC, DONT, option]) (acknowledge)
No guard needed — WONT is always safe to acknowledge
[ ] _handle_do(self, option: int) -> None:
Server requests we enable an option
Accepted options (respond WILL):
- OPT_TTYPE — we will send terminal type; also queue TTYPE subnegotiation response
- OPT_NAWS — we will send window size; also queue NAWS subnegotiation response
Refused options (respond WONT):
- Everything else
Update _option_state[option]["local"] = True for accepted
Append response to _response_buffer
Guard: Same loop-prevention as _handle_will
[ ] _handle_dont(self, option: int) -> None:
Server requests we disable an option
Update _option_state[option]["local"] = False
Respond bytes([IAC, WONT, option]) (acknowledge)
[ ] _queue_ttype_response(self) -> None:
Build and append TTYPE IS subnegotiation: IAC SB TTYPE IS "TELTEST" IAC SE
Bytes: bytes([IAC, SB, OPT_TTYPE, TTYPE_IS]) + b"TELTEST" + bytes([IAC, SE])
[ ] _queue_naws_response(self, width: int = 80, height: int = 24) -> None:
Build and append NAWS subnegotiation: IAC SB NAWS <w_hi> <w_lo> <h_hi> <h_lo> IAC SE
Width/height encoded as two big-endian bytes each
Must escape any 0xFF bytes within the payload (replace 0xFF with 0xFF 0xFF)

1.2.4 Subnegotiation Handler¶

[ ] _handle_subnegotiation(self, option: int, data: bytes) -> None:
Dispatch based on option:
- OPT_TTYPE: If data starts with TTYPE_SEND, queue _queue_ttype_response()
- OPT_GMCP: Call _handle_gmcp_subnegotiation(data)
- All others: ignore (log at debug level if needed)
[ ] _handle_gmcp_subnegotiation(self, data: bytes) -> None:
Parse GMCP payload: <package_name> <json_data> (space-separated)
Split on first space: package = data[:space_idx].decode("utf-8"), json_str = data[space_idx+1:]
If no space found, package = data.decode("utf-8"), json_data = {}
Parse JSON via json.loads(json_str) — on json.JSONDecodeError, store raw string as {"_raw": json_str.decode("utf-8")}
Append (package, json_data) to self._gmcp_messages

1.2.5 GMCP Sending Helper¶

[ ] build_gmcp_message(self, package: str, data: dict[str, Any]) -> bytes:
Build a GMCP subnegotiation for sending: IAC SB GMCP <package> <json> IAC SE
json_payload = json.dumps(data, separators=(",", ":")) (compact)
payload = package.encode("utf-8") + b" " + json_payload.encode("utf-8")
Return bytes([IAC, SB, OPT_GMCP]) + payload + bytes([IAC, SE])
Does not append to _response_buffer — returned directly for send_raw()

1.2.6 Unit Tests¶

File: packages/teltest/tests/test_protocol.py

[ ] test_plain_text_passes_through — Feed ASCII text with no IAC; feed() returns identical bytes
[ ] test_escaped_iac_becomes_literal — Feed IAC IAC; feed() returns single 0xFF byte
[ ] test_will_echo_responds_do_echo — Feed IAC WILL ECHO; get_response() returns IAC DO ECHO
[ ] test_will_sga_responds_do_sga — Feed IAC WILL SGA; get_response() returns IAC DO SGA
[ ] test_will_mccp2_responds_dont_mccp2 — Feed IAC WILL MCCP2; get_response() returns IAC DONT MCCP2
[ ] test_will_msdp_responds_dont_msdp — Feed IAC WILL MSDP; get_response() returns IAC DONT MSDP
[ ] test_will_mxp_responds_dont_mxp — Feed IAC WILL MXP; get_response() returns IAC DONT MXP
[ ] test_will_unknown_option_responds_dont — Feed IAC WILL 99; get_response() returns IAC DONT 99
[ ] test_wont_responds_dont — Feed IAC WONT ECHO; get_response() returns IAC DONT ECHO
[ ] test_do_ttype_responds_will_ttype_and_subneg — Feed IAC DO TTYPE; get_response() includes IAC WILL TTYPE; then feed IAC SB TTYPE SEND IAC SE; get_response() includes IAC SB TTYPE IS TELTEST IAC SE
[ ] test_do_naws_responds_will_naws_and_subneg — Feed IAC DO NAWS; get_response() includes IAC WILL NAWS and an NAWS subnegotiation with 80×24
[ ] test_do_unknown_option_responds_wont — Feed IAC DO 99; get_response() returns IAC WONT 99
[ ] test_dont_responds_wont — Feed IAC DONT TTYPE; get_response() returns IAC WONT TTYPE
[ ] test_ga_sets_flag — Feed IAC GA; consume_ga() returns True, subsequent call returns False
[ ] test_nop_is_ignored — Feed IAC NOP; feed() returns empty bytes, no response queued
[ ] test_iac_sequences_stripped_from_text — Feed b"Hello" + IAC WILL ECHO + b"World"; feed() returns b"HelloWorld"
[ ] test_mixed_text_and_commands — Feed interleaved text and multiple IAC sequences; verify only clean text returned
[ ] test_partial_iac_at_buffer_boundary — Feed b"text" + bytes([IAC]) in one call, then bytes([WILL, OPT_ECHO]) in second call; verify text from first call, response from second
[ ] test_subnegotiation_basic — Feed IAC SB <option> <data> IAC SE; verify data is consumed, no text output
[ ] test_subnegotiation_with_escaped_iac — Feed SB containing IAC IAC (escaped 0xFF); verify the 0xFF is preserved in SB data
[ ] test_gmcp_disabled_responds_dont — Protocol with negotiate_gmcp=False; feed IAC WILL GMCP; response is IAC DONT GMCP
[ ] test_gmcp_enabled_responds_do — Protocol with negotiate_gmcp=True; feed IAC WILL GMCP; response is IAC DO GMCP
[ ] test_gmcp_message_extraction — Feed GMCP subnegotiation IAC SB GMCP "Char.Vitals {"hp":100}" IAC SE; get_gmcp_messages() returns [("Char.Vitals", {"hp": 100})]
[ ] test_gmcp_message_no_json_body — Feed GMCP subneg with package name only (no space, no JSON); get_gmcp_messages() returns [("PackageName", {})]
[ ] test_gmcp_message_invalid_json — Feed GMCP subneg with invalid JSON; result stores raw string in {"_raw": ...}
[ ] test_build_gmcp_message — Verify build_gmcp_message("Char.Login", {"name": "Test"}) produces correct byte sequence
[ ] test_negotiation_loop_prevention — Feed IAC WILL ECHO twice; only one IAC DO ECHO in response (second is suppressed)
[ ] test_option_state_tracking — Negotiate ECHO; is_option_active(OPT_ECHO) returns True; feed IAC WONT ECHO; returns False
[ ] test_naws_escapes_0xff_in_payload — Feed IAC DO NAWS with protocol configured for width 255; verify NAWS subneg payload escapes the 0xFF byte
[ ] test_empty_feed — Feed empty bytes; returns empty bytes, no response
[ ] test_feed_only_iac_sequences — Feed only IAC commands with no text; returns empty bytes

Section 1.3: ANSIStripper — Escape Code Removal¶

Package: teltest | Priority: P0 | Dependencies: 1.1

File: packages/teltest/src/teltest/ansi.py

1.3.1 ANSIStripper Class¶

[ ] class ANSIStripper:
[ ] ANSI_PATTERN: re.Pattern[str] — class-level compiled regex
- Pattern: r"\x1b\[[0-9;]*[A-Za-z]|\x1b\][^\x07]*(?:\x07|\x1b\\)|\x1b[()][AB012]|\x1b[>=<]?"
- Covers:
- CSI sequences: ESC [ <params> <letter> (colors, cursor movement, etc.)
- OSC sequences: ESC ] ... BEL or ESC ] ... ESC \ (window title, etc.)
- Character set selectors: ESC ( B, ESC ) 0, etc.
- Simple escapes: ESC >, ESC =, ESC <
[ ] __init__(self) -> None:
- self._partial: str = "" — stores partial escape sequence split across chunks
[ ] strip(self, text: str) -> str:
- Prepend self._partial to text
- If text ends with \x1b or with \x1b[ followed by digits/semicolons (but no terminating letter), save the incomplete sequence to self._partial and remove from text
- Apply ANSI_PATTERN.sub("", text) to remove complete sequences
- Return cleaned text
[ ] reset(self) -> None:
- Clear self._partial

1.3.2 Module-Level Convenience Function¶

[ ] def strip_ansi(text: str) -> str:
Stateless single-call version (no partial tracking)
return ANSIStripper.ANSI_PATTERN.sub("", text)

1.3.3 Unit Tests¶

File: packages/teltest/tests/test_ansi.py

[ ] test_plain_text_unchanged — strip_ansi("Hello World") returns "Hello World"
[ ] test_strip_sgr_color_codes — strip_ansi("\x1b[31mRed\x1b[0m") returns "Red"
[ ] test_strip_bold — strip_ansi("\x1b[1mBold\x1b[0m") returns "Bold"
[ ] test_strip_256_color — strip_ansi("\x1b[38;5;196mRed\x1b[0m") returns "Red"
[ ] test_strip_24bit_color — strip_ansi("\x1b[38;2;255;0;0mRed\x1b[0m") returns "Red"
[ ] test_strip_cursor_movement — strip_ansi("\x1b[2J\x1b[HHello") returns "Hello"
[ ] test_strip_multiple_sequences — Text with several interleaved codes; all removed, text preserved
[ ] test_empty_string — strip_ansi("") returns ""
[ ] test_no_escape_codes — ASCII text with special chars ([], ;) but no ESC; unchanged
[ ] test_osc_sequence_with_bel — strip_ansi("\x1b]0;Window Title\x07Text") returns "Text"
[ ] test_osc_sequence_with_st — strip_ansi("\x1b]0;Title\x1b\\Text") returns "Text"
[ ] test_strip_preserves_newlines — strip_ansi("\x1b[32mLine1\r\nLine2\x1b[0m") returns "Line1\r\nLine2"
[ ] test_stateful_partial_sequence_across_chunks — ANSIStripper instance: strip("\x1b[31mHello\x1b") returns "Hello", then strip("[0mWorld") returns "World"
[ ] test_stateful_partial_csi_params_across_chunks — strip("Text\x1b[38;5") returns "Text", then strip(";196mMore") returns "More"
[ ] test_stateful_reset_clears_partial — After partial, reset(), then new text with ] is not treated as continuation
[ ] test_only_escape_codes — strip_ansi("\x1b[0m\x1b[31m\x1b[0m") returns ""
[ ] test_realistic_mud_output — A line like "\x1b[1;33m[HP:100]\x1b[0m \x1b[32mYou are standing...\x1b[0m" returns "[HP:100] You are standing..."

Section 1.4: OutputBuffer — Line Splitting, Prompt Detection, Ring Buffer¶

Package: teltest | Priority: P0 | Dependencies: 1.1

File: packages/teltest/src/teltest/buffer.py

The OutputBuffer is the central data structure that stores received text, splits it into lines, detects prompts via multiple signals, and tracks a read cursor for cursor-based expect() matching.

1.4.1 OutputBuffer Class¶

[ ] class OutputBuffer:
[ ] __init__(self, *, max_lines: int = 10_000, prompt_pattern: re.Pattern[str] | None = None, prompt_settle_time: float = 0.3) -> None:
- self._lines: collections.deque[str] = collections.deque(maxlen=max_lines) — ring buffer of complete lines
- self._partial: str = "" — current unterminated line (potential prompt)
- self._cursor: int = 0 — read cursor position (index into logical character stream)
- self._total_chars: int = 0 — total characters ever appended (monotonic)
- self._text: str = "" — full buffer text (rebuilt on access, cached)
- self._text_dirty: bool = True — invalidate cache on append
- self._prompt_pattern: re.Pattern[str] | None = prompt_pattern
- self._prompt_settle_time: float = prompt_settle_time
- self._ga_received: bool = False — GA signal forwarded from protocol
- self._last_data_time: float = 0.0 — monotonic time of last append() call
- self._new_data: asyncio.Event = asyncio.Event() — signaled on new data arrival
- self._closed: bool = False — set on EOF
- self._max_lines: int = max_lines
[ ] append(self, text: str) -> None:
- Split text on \r\n and \n (handle both; normalize \r\n to \n internally)
- For each complete line: append to self._lines, increment self._total_chars by len(line) + 1 (for newline)
- Any trailing non-newline-terminated text goes into self._partial
- If self._partial was non-empty before this call and new text starts without newline, concatenate
- Update self._last_data_time = time.monotonic()
- Set self._text_dirty = True
- Call self._new_data.set() to wake up any waiting expect()
[ ] signal_ga(self) -> None:
- self._ga_received = True
- Call self._new_data.set() to wake up prompt waiters
[ ] signal_closed(self) -> None:
- self._closed = True
- Call self._new_data.set() to unblock any waiting expect() with EOF
[ ] @property text(self) -> str:
- If self._text_dirty: rebuild self._text from "\n".join(self._lines) + ("\n" + self._partial if partial exists, or "\n" if lines exist and partial is empty)
- Return self._text
[ ] @property lines(self) -> list[str]:
- Return list(self._lines)
[ ] @property partial_line(self) -> str:
- Return self._partial (current unterminated text)
[ ] @property is_closed(self) -> bool:
- Return self._closed

1.4.2 Cursor Management¶

[ ] @property cursor(self) -> int:
Return self._cursor
[ ] advance_cursor(self) -> None:
Set self._cursor to current end of buffer (len(self.text))
Used by send() to skip past all currently buffered text
[ ] advance_cursor_past(self, position: int) -> None:
Set self._cursor = position
Used by expect() to advance cursor past the match
[ ] text_from_cursor(self) -> str:
Return self.text[self._cursor:]
Text received since last cursor advance (what expect() searches)
[ ] reset(self) -> None:
Clear self._lines, self._partial, reset self._cursor = 0, self._total_chars = 0
Set self._text_dirty = True
Clear self._ga_received
Clear self._new_data event

1.4.3 Prompt Detection¶

[ ] detect_prompt(self) -> tuple[bool, PromptSignal | None, str]:
Returns (is_prompt, signal, prompt_text) tuple
Signal 1 — GA: If self._ga_received is True and self._partial is non-empty:
- self._ga_received = False
- Return (True, PromptSignal.GA, self._partial)
Signal 2 — Regex: If self._prompt_pattern is not None and self._partial is non-empty:
- If self._prompt_pattern.search(self._partial) matches:
- Return (True, PromptSignal.REGEX, self._partial)
Signal 3 — Quiescence: If self._partial is non-empty:
- If time.monotonic() - self._last_data_time >= self._prompt_settle_time:
- Return (True, PromptSignal.QUIESCENCE, self._partial)
Otherwise: (False, None, "")

1.4.4 Async Waiting¶

[ ] async def wait_for_data(self, timeout: float) -> bool:
Wait for self._new_data event with timeout
Clear the event after waking
Returns True if new data arrived, False on timeout
Raises nothing — caller checks self._closed separately
Implementation: wrap self._new_data.wait() with asyncio.wait_for(..., timeout), catch asyncio.TimeoutError → return False, then clear event

1.4.5 Unit Tests¶

File: packages/teltest/tests/test_buffer.py

[ ] test_append_complete_lines — Append "Line1\r\nLine2\r\n"; lines is ["Line1", "Line2"], partial_line is ""
[ ] test_append_partial_line — Append "Hello"; lines is [], partial_line is "Hello"
[ ] test_append_mixed — Append "Line1\r\nPartial"; lines is ["Line1"], partial_line is "Partial"
[ ] test_append_continues_partial — Append "Hel", then "lo\r\n"; lines is ["Hello"], partial_line is ""
[ ] test_append_unix_newlines — Append "A\nB\n"; lines is ["A", "B"]
[ ] test_append_crlf_normalization — Append "A\r\nB\r\n"; lines is ["A", "B"]
[ ] test_text_property_full_content — Append lines and partial; text contains all concatenated with newlines
[ ] test_ring_buffer_max_lines — Create buffer with max_lines=3; append 5 lines; only last 3 in lines
[ ] test_ring_buffer_drops_oldest — Verify first lines are dropped when capacity exceeded
[ ] test_cursor_starts_at_zero — New buffer has cursor == 0
[ ] test_advance_cursor — Append text, advance_cursor(); text_from_cursor() returns ""; append more; text_from_cursor() returns new text only
[ ] test_advance_cursor_past — Append "ABCDEF"; advance_cursor_past(3); text_from_cursor() starts at char index 3
[ ] test_text_from_cursor_after_send — Simulates send scenario: append text, advance cursor, append response; text_from_cursor() only shows response
[ ] test_reset_clears_everything — Append text, reset(); text is "", lines is [], cursor is 0
[ ] test_signal_ga_sets_flag — signal_ga(); detect_prompt() returns (True, PromptSignal.GA, partial) when partial is non-empty
[ ] test_detect_prompt_ga_signal — Append "> " (partial); signal_ga(); detect_prompt() returns GA signal with "> "
[ ] test_detect_prompt_ga_clears_after_read — After detect_prompt() returns GA, second call does not return GA again (if no new GA)
[ ] test_detect_prompt_regex_signal — Create buffer with prompt_pattern=re.compile(r"^> $"); append "> "; detect_prompt() returns REGEX signal
[ ] test_detect_prompt_regex_no_match — Partial text "some text" does not match prompt regex; returns (False, None, "")
[ ] test_detect_prompt_quiescence — Append "> ", wait for prompt_settle_time; detect_prompt() returns QUIESCENCE signal
[ ] test_detect_prompt_quiescence_too_soon — Append "> " immediately; detect_prompt() returns (False, None, "") (not enough time elapsed)
[ ] test_detect_prompt_priority_ga_over_regex — Both GA flag set and regex matches; GA is returned (priority 1)
[ ] test_detect_prompt_empty_partial — No partial text; detect_prompt() returns (False, None, "") regardless of GA flag
[ ] test_signal_closed — signal_closed(); is_closed returns True
[ ] test_wait_for_data_returns_true — Append data from a background task; wait_for_data(1.0) returns True
[ ] test_wait_for_data_timeout — No data appended; wait_for_data(0.1) returns False
[ ] test_wait_for_data_unblocked_by_close — signal_closed() from background task; wait_for_data(5.0) returns True (event was set)
[ ] test_wait_for_data_unblocked_by_ga — signal_ga() from background task; wait_for_data(5.0) returns True
[ ] test_new_data_event_cleared_after_wait — After wait_for_data() returns, event is cleared; subsequent wait_for_data(0.01) returns False if no new data
[ ] test_append_empty_string — Append ""; no change to buffer state
[ ] test_multiple_appends_accumulate — Multiple append() calls; text contains all data in order

Section 1.5: MUDClient — Connection Lifecycle & Background Reader¶

Package: teltest | Priority: P0 | Dependencies: 1.2, 1.3, 1.4

File: packages/teltest/src/teltest/client.py

The MUDClient is the top-level class that ties together TelnetProtocol, ANSIStripper, and OutputBuffer with TCP connection management and a background reader task.

1.5.1 Test Fixtures¶

File: packages/teltest/tests/conftest.py

[ ] echo_server fixture — An async TCP server that echoes back whatever it receives:

@pytest.fixture
async def echo_server() -> AsyncGenerator[tuple[str, int], None]:
    """Start a TCP server on an ephemeral port that echoes received data."""

Binds to ("127.0.0.1", 0) to get a free port
For each connection: reads data, writes it back
Yields (host, port) tuple
Cleans up server on teardown

[ ] telnet_server fixture — An async TCP server that sends telnet IAC negotiation:

@pytest.fixture
async def telnet_server() -> AsyncGenerator[TelnetServerHelper, None]:
    """Start a TCP server that speaks telnet protocol."""

TelnetServerHelper class with:
- host: str — server bind address
- port: int — server bind port
- send(data: bytes) -> None — send raw bytes to connected client
- send_line(text: str) -> None — send text with \r\n appended
- send_iac(command: int, option: int) -> None — send 3-byte IAC sequence
- send_ga() -> None — send bytes([IAC, GA])
- send_gmcp(package: str, data: dict[str, Any]) -> None — send GMCP subnegotiation
- received: list[bytes] — bytes received from client
- close() -> None — close the client connection (simulate server crash/disconnect)
- async wait_for_connection(timeout: float = 2.0) -> None — wait for a client to connect
Binds to ephemeral port
Accepts one connection at a time
Yields the helper
Cleans up on teardown

[ ] silent_server fixture — A TCP server that accepts connections but sends nothing:

@pytest.fixture
async def silent_server() -> AsyncGenerator[tuple[str, int], None]:
    """TCP server that accepts connections but sends no data."""

[ ] closing_server fixture — A TCP server that accepts and immediately closes:

@pytest.fixture
async def closing_server() -> AsyncGenerator[tuple[str, int], None]:
    """TCP server that accepts connections then immediately closes them."""

1.5.2 MUDClient Class¶

[ ] class MUDClient:
[ ] __init__(self, host: str = "localhost", port: int = 4000, *, timeout: float = 5.0, strip_ansi: bool = True, encoding: str = "utf-8", negotiate_gmcp: bool = False, prompt_pattern: re.Pattern[str] | None = None, prompt_settle_time: float = 0.3, max_history_lines: int = 10_000) -> None:
- self._host: str = host
- self._port: int = port
- self._timeout: float = timeout
- self._strip_ansi: bool = strip_ansi
- self._encoding: str = encoding
- self._protocol: TelnetProtocol = TelnetProtocol(negotiate_gmcp=negotiate_gmcp)
- self._ansi_stripper: ANSIStripper = ANSIStripper() (only used if strip_ansi)
- self._buffer: OutputBuffer = OutputBuffer(max_lines=max_history_lines, prompt_pattern=prompt_pattern, prompt_settle_time=prompt_settle_time)
- self._reader: asyncio.StreamReader | None = None
- self._writer: asyncio.StreamWriter | None = None
- self._reader_task: asyncio.Task[None] | None = None
- self._reader_error: BaseException | None = None — last exception from reader task
- self._connected: bool = False
- self._transcript: collections.deque[TranscriptEntry] = collections.deque(maxlen=max_history_lines)
- self._gmcp_messages: list[tuple[str, dict[str, Any]]] = [] — accumulated GMCP messages

1.5.3 Connection Lifecycle¶

[ ] async def connect(self) -> None:
Open TCP connection: self._reader, self._writer = await asyncio.wait_for(asyncio.open_connection(self._host, self._port), timeout=self._timeout)
Set self._connected = True
Clear self._reader_error = None
Start background reader: self._reader_task = asyncio.create_task(self._reader_loop(), name="teltest-reader")
Attach error handler: self._reader_task.add_done_callback(self._on_reader_done)
[ ] async def disconnect(self) -> None:
If not connected, return silently (idempotent)
Set self._connected = False
Cancel the reader task: self._reader_task.cancel()

Await the reader task with suppressed CancelledError:

with contextlib.suppress(asyncio.CancelledError):
    await self._reader_task

Close the writer: self._writer.close(), await self._writer.wait_closed()
Signal buffer closed: self._buffer.signal_closed()
Set self._reader = None, self._writer = None, self._reader_task = None
[ ] async def __aenter__(self) -> "MUDClient":
await self.connect()
return self
[ ] async def __aexit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: object) -> None:
await self.disconnect()
[ ] @property is_connected(self) -> bool:
return self._connected

1.5.4 Background Reader Task¶

[ ] async def _reader_loop(self) -> None:

Continuous loop reading from self._reader:

while self._connected:
    data = await self._reader.read(4096)
    if not data:
        # EOF — server closed connection
        self._buffer.signal_closed()
        self._connected = False
        return
    # Process through telnet protocol layer
    clean_bytes = self._protocol.feed(data)
    # Send any negotiation responses back to server
    response = self._protocol.get_response()
    if response:
        self._writer.write(response)
        await self._writer.drain()
    # Check for Go-Ahead signal
    if self._protocol.consume_ga():
        self._buffer.signal_ga()
    # Extract GMCP messages
    for package, gmcp_data in self._protocol.get_gmcp_messages():
        self._gmcp_messages.append((package, gmcp_data))
        self._transcript.append(TranscriptEntry(
            timestamp=time.monotonic(),
            direction="gmcp_recv",
            text=f"{package} {json.dumps(gmcp_data)}",
        ))
    # Decode bytes to text and optionally strip ANSI
    text = clean_bytes.decode(self._encoding, errors="replace")
    if self._strip_ansi:
        text = self._ansi_stripper.strip(text)
    # Append to output buffer
    if text:
        self._buffer.append(text)
        self._transcript.append(TranscriptEntry(
            timestamp=time.monotonic(),
            direction="recv",
            text=text,
        ))

Exception handling: The loop does NOT catch exceptions. They propagate to the task's result, handled by _on_reader_done.
[ ] def _on_reader_done(self, task: asyncio.Task[None]) -> None:
Callback attached via add_done_callback on the reader task
If task.cancelled(): do nothing (expected during disconnect())
If task.exception() is not None:
- Store: self._reader_error = task.exception()
- Set: self._connected = False
- Signal buffer: self._buffer.signal_closed()

1.5.5 Sending¶

[ ] async def send(self, text: str) -> None:
Raise NotConnected if not self._connected
Encode: data = (text + "\r\n").encode(self._encoding)
Write: self._writer.write(data), await self._writer.drain()
Advance read cursor: self._buffer.advance_cursor()
Record transcript: self._transcript.append(TranscriptEntry(timestamp=time.monotonic(), direction="send", text=text))
[ ] async def send_raw(self, data: bytes) -> None:
Raise NotConnected if not self._connected
Write: self._writer.write(data), await self._writer.drain()
Do NOT advance cursor (raw sends are for protocol-level interaction)
Record transcript with direction="send", text=repr(data)
[ ] async def send_gmcp(self, package: str, data: dict[str, Any]) -> None:
Raise NotConnected if not self._connected
Build message: msg = self._protocol.build_gmcp_message(package, data)
Write: self._writer.write(msg), await self._writer.drain()
Record transcript: TranscriptEntry(timestamp=..., direction="gmcp_send", text=f"{package} {json.dumps(data)}")

1.5.6 Buffer Access Properties¶

[ ] @property output(self) -> str:
return self._buffer.text
[ ] @property recent(self) -> str:
return self._buffer.text_from_cursor()
[ ] def clear_buffer(self) -> None:
self._buffer.reset()
[ ] @property transcript(self) -> list[TranscriptEntry]:
return list(self._transcript)
[ ] @property reader_error(self) -> BaseException | None:
return self._reader_error
[ ] @property gmcp_messages(self) -> list[tuple[str, dict[str, Any]]]:
return list(self._gmcp_messages)

1.5.7 Reader Error Re-raise Contract¶

The design specifies that reader exceptions are captured and re-raised in the next expect() call. Since expect() is implemented in Phase 2 (Expectation Engine), MUDClient exposes the mechanism for it:

[ ] def _check_reader_health(self) -> None:
Called at the start of every expect*() method (Phase 2 will call this)
If self._reader_error is not None:
- Raise ConnectionClosed(buffer=self._buffer.text) from self._reader_error
If self._buffer.is_closed and not self._connected:
- Raise ConnectionClosed(buffer=self._buffer.text)

1.5.8 Unit Tests¶

File: packages/teltest/tests/test_client.py

Connection lifecycle tests:

[ ] test_connect_opens_tcp_connection — Connect to echo_server; is_connected returns True
[ ] test_connect_starts_reader_task — Connect; _reader_task is not None and not done
[ ] test_disconnect_closes_connection — Connect then disconnect; is_connected returns False
[ ] test_disconnect_cancels_reader_task — Connect then disconnect; _reader_task is None (cleaned up)
[ ] test_disconnect_idempotent — Disconnect twice; no error on second call
[ ] test_connect_timeout_raises — Connect to non-listening port with short timeout; raises asyncio.TimeoutError or ConnectionRefusedError
[ ] test_context_manager_connect_disconnect — async with MUDClient(host, port) as c: — is_connected is True inside, False after
[ ] test_context_manager_disconnect_on_exception — Exception inside async with; client still disconnects cleanly

Send tests:

[ ] test_send_appends_crlf — Send "hello" to telnet_server; server receives b"hello\r\n"
[ ] test_send_advances_cursor — Connect to telnet_server; server sends text; after brief wait, send("cmd"); cursor is at end of pre-send buffer
[ ] test_send_raw_no_crlf — send_raw(b"raw") to telnet_server; server receives exactly b"raw"
[ ] test_send_raw_does_not_advance_cursor — send_raw(b"data"); cursor unchanged from before call
[ ] test_send_when_disconnected_raises — Disconnect then send("x"); raises NotConnected
[ ] test_send_records_transcript — Send text; transcript contains entry with direction="send" and the sent text

Reader task tests:

[ ] test_reader_receives_plain_text — telnet_server sends "Hello\r\n"; after short wait, output contains "Hello"
[ ] test_reader_handles_iac_negotiation — telnet_server sends IAC WILL ECHO; server receives IAC DO ECHO response; output does not contain IAC bytes
[ ] test_reader_strips_ansi_when_enabled — telnet_server sends "\x1b[31mRed\x1b[0m\r\n"; output contains "Red" (stripped)
[ ] test_reader_preserves_ansi_when_disabled — MUDClient with strip_ansi=False; telnet_server sends ANSI; output contains escape codes
[ ] test_reader_detects_eof — telnet_server closes connection; after short wait, is_connected is False and _buffer.is_closed is True
[ ] test_reader_error_stored_on_exception — Force an exception scenario in reader; reader_error is set to the exception
[ ] test_reader_eof_sets_connection_closed — Server closes connection; _check_reader_health() raises ConnectionClosed
[ ] test_reader_error_reraised_as_connection_closed — Reader encounters error; _check_reader_health() raises ConnectionClosed with __cause__ set to the original
[ ] test_reader_records_recv_transcript — telnet_server sends text; transcript contains direction="recv" entries with the text

GMCP tests:

[ ] test_send_gmcp_builds_correct_bytes — send_gmcp("Char.Login", {"name": "X"}); server receives valid GMCP subnegotiation bytes
[ ] test_reader_extracts_gmcp_messages — telnet_server sends GMCP subneg after GMCP negotiation; gmcp_messages contains extracted (package, data) tuple
[ ] test_gmcp_disabled_no_negotiation — MUDClient with negotiate_gmcp=False; server sends IAC WILL GMCP; server receives IAC DONT GMCP

Buffer access tests:

[ ] test_output_returns_all_text — Server sends multiple lines; output contains all of them
[ ] test_recent_returns_text_since_cursor — Server sends text; send("cmd"); server sends more; recent only contains post-send text
[ ] test_clear_buffer_resets_all — Send text, clear_buffer(); output is "", recent is ""

Multi-line and timing tests:

[ ] test_multiple_lines_buffered — Server sends 5 lines rapidly; all appear in output
[ ] test_partial_line_buffered_as_prompt — Server sends "> " without newline; appears in _buffer.partial_line
[ ] test_transcript_ordering — Mix of sends and receives; transcript entries are in chronological order

Edge case tests:

[ ] test_connect_to_closing_server — Connect to closing_server; reader detects EOF; is_connected becomes False
[ ] test_large_data_read — Server sends 100KB of text; all data received without error
[ ] test_encoding_error_replaced — Server sends invalid UTF-8 bytes (after IAC stripping); decoded with replacement chars, no crash

Section 1.6: Integration Tests¶

Package: teltest | Priority: P0 | Dependencies: 1.2, 1.3, 1.4, 1.5

File: packages/teltest/tests/test_integration.py

These tests verify the full stack (protocol → ANSI → buffer → client) working together end-to-end.

[ ] test_full_telnet_handshake_and_text — telnet_server sends IAC WILL ECHO, IAC WILL SGA, IAC DO TTYPE, IAC DO NAWS, then text lines; client auto-negotiates all and receives clean text
[ ] test_gmcp_roundtrip — Client with negotiate_gmcp=True; server sends IAC WILL GMCP, then a GMCP message; client extracts message; client sends GMCP back; server receives correct bytes
[ ] test_prompt_detection_via_ga — Server sends "Hello\r\n> " then IAC GA; buffer detect_prompt() returns (True, PromptSignal.GA, "> ")
[ ] test_prompt_detection_via_regex — Client with prompt_pattern=re.compile(r"> $"); server sends "> "; after settle time, detect_prompt() returns regex signal
[ ] test_ansi_stripped_before_buffering — Server sends ANSI-colored text interspersed with IAC sequences; output contains only clean text
[ ] test_interleaved_iac_and_text — Server sends text, IAC sequences, more text, GMCP, more text in one large TCP write; all text arrives clean, GMCP extracted, negotiation responded to
[ ] test_multiple_rapid_sends — Client sends 10 commands rapidly; all appear in server's received data with \r\n terminators
[ ] test_server_disconnect_during_read — Server sends partial text then closes; client detects EOF, is_connected is False, partial text preserved in buffer

Dependency Graph¶

1.1 Package Scaffolding & Shared Types
 │
 ├── 1.2 TelnetProtocol (depends on 1.1 for constants/types)
 │
 ├── 1.3 ANSIStripper (depends on 1.1 for package structure)
 │
 ├── 1.4 OutputBuffer (depends on 1.1 for PromptSignal, asyncio.Event)
 │
 └── 1.5 MUDClient (depends on 1.2 + 1.3 + 1.4)
      │
      └── 1.6 Integration Tests (depends on 1.5)

Sections 1.2, 1.3, and 1.4 can be implemented in parallel since they have no cross-dependencies. Section 1.5 requires all three. Section 1.6 requires 1.5.

Verification Checklist¶

After all sections complete:

[ ] uv run pytest packages/teltest/tests/ -v — all tests pass
[ ] uv run mypy packages/teltest/src/teltest/ --strict — no type errors
[ ] uv run ruff check packages/teltest/ — no lint errors
[ ] Zero MAID imports anywhere in packages/teltest/
[ ] All public API types exported from packages/teltest/src/teltest/__init__.py
[ ] packages/teltest/pyproject.toml has zero runtime dependencies