Expectation Engine — Implementation Plan¶

Design Document: docs/designs/teltest/README.md §2 (Expectation Engine), §API Design (MUDClient API) Priority: P0 — Core framework capability; all E2E assertions depend on this Estimated Duration: 4 weeks across 6 sections (2.1–2.6)

Summary¶

This plan implements the TelTest Expectation Engine — the pattern-matching assertion layer that powers every expect() call in the E2E testing framework. The expectation engine is the primary mechanism by which tests verify server output, and every other TelTest component (fixtures, script DSL, session recording) depends on it.

The implementation adds:

Match dataclass — frozen result object returned by successful expect() calls with matched text, full line, pattern, elapsed time, and buffer context
ExpectTimeout exception — rich AssertionError subclass with full buffer dump, transcript excerpt, and diagnostic hints
ConnectionClosed exception — raised on server disconnect during expect() with buffer context
UnexpectedMatch exception — raised by expect_not() when a forbidden pattern appears before the positive boundary
TranscriptEntry dataclass — timestamped record of send/recv/gmcp events for debugging
Expectation methods — expect(), expect_prompt(), expect_line(), expect_sequence(), expect_any(), expect_not() with cursor-based matching, configurable timeouts, and timeout multiplier support

Key design decisions from the design doc:

Cursor-based matching — Each expect() and send() advances a read cursor so assertions only match text received since the last interaction. This eliminates an entire class of flaky tests where stale output matches new patterns. (Design doc §2, post-review)
from_start=True opt-in — Full-buffer scan is available but never the default
expect_not() requires until — A positive boundary parameter avoids unconditional sleeps. All three adversarial reviewers flagged the original sleep-based design. (Design doc Appendix C, finding #4)
Multi-signal prompt detection — expect_prompt() uses GA signal, regex matching, and quiescence detection to avoid TCP fragmentation race conditions. (Design doc §2, finding #6)
Timeout multiplier — All timeouts scale via TELTEST_TIMEOUT_MULTIPLIER for CI environments (Design doc §Configuration)

Existing infrastructure leveraged:

OutputBuffer at packages/teltest/src/teltest/buffer.py (Phase 1) — line splitting, prompt tracking, ring buffer
TelnetProtocol at packages/teltest/src/teltest/protocol.py (Phase 1) — IAC negotiation, GA signal detection
MUDClient at packages/teltest/src/teltest/client.py (Phase 1) — connection lifecycle, send(), background reader task, _cursor position, _new_data event

Prerequisites (must be complete before this phase):

Phase 1 (MUDClient core) — connect(), disconnect(), send(), send_raw(), background reader task, OutputBuffer, TelnetProtocol, ANSIStripper
OutputBuffer must expose: get_text(from_pos) -> str, get_line_at(pos) -> str, current_length -> int, get_lines(start, count) -> list[str], has_unterminated_line -> bool, get_unterminated_line -> str
MUDClient must expose: _cursor: int, _new_data: asyncio.Event, _buffer: OutputBuffer, _connected: bool, _reader_error: Exception | None, _timeout: float

2.1 Data Models¶

Package: teltest | Priority: P0 | Dependencies: Phase 1 (OutputBuffer)

2.1.1 `Match` Dataclass¶

[ ] Create Match in packages/teltest/src/teltest/expect.py:

@dataclass(frozen=True)
class Match:
    """Result of a successful expect() call."""
    text: str            # The matched text (substring or regex group 0)
    line: str            # The full line containing the match
    pattern: str         # String representation of the pattern that matched
    elapsed: float       # Seconds waited before match was found
    buffer_before: str   # Buffer context before the match (last N lines)

text — For substring matches, the literal substring found. For regex matches, the full match (re.Match.group(0)).
line — The complete line (delimited by \r\n or \n) containing the match. For multi-line regex matches, this is the first line of the match.
pattern — String representation: for str patterns, the literal string; for re.Pattern, the .pattern attribute.
elapsed — time.monotonic() delta from when the expect call started to when the match was found.
buffer_before — The last _CONTEXT_LINES (default 20) lines of buffer content preceding the match position, for debugging context.
[ ] Add module-level constant _CONTEXT_LINES: int = 20 controlling how many lines of context buffer_before captures.
[ ] Implement __str__ on Match returning a human-readable summary: "Matched '{self.text}' on line '{self.line}' (pattern='{self.pattern}', elapsed={self.elapsed:.3f}s)"

2.1.2 `TranscriptEntry` Dataclass¶

[ ] Create TranscriptEntry in packages/teltest/src/teltest/expect.py:

@dataclass(frozen=True)
class TranscriptEntry:
    """Single entry in the session transcript."""
    timestamp: float                                              # time.monotonic() value
    direction: Literal["send", "recv", "gmcp_send", "gmcp_recv"] # Event type
    text: str                                                     # Content sent/received

timestamp is absolute time.monotonic() recorded at event time; display code computes relative offsets from the first entry
text for recv entries is post-ANSI-stripping (if stripping is enabled) and post-IAC-filtering
text for gmcp_send/gmcp_recv is the JSON-serialized GMCP payload prefixed with the package name: "package.name {json}"

2.1.3 `Transcript` Recorder¶

[ ] Create Transcript class in packages/teltest/src/teltest/expect.py:

class Transcript:
    """Ring-buffer transcript of session events."""

    def __init__(self, max_entries: int = 10_000) -> None: ...

    def record_send(self, text: str) -> None: ...
    def record_recv(self, text: str) -> None: ...
    def record_gmcp_send(self, package: str, data: dict[str, Any]) -> None: ...
    def record_gmcp_recv(self, package: str, data: dict[str, Any]) -> None: ...

    @property
    def entries(self) -> list[TranscriptEntry]: ...

    def recent(self, n: int = 5) -> list[TranscriptEntry]: ...

    def format_recent(self, n: int = 5, relative_to: float | None = None) -> str: ...

Uses collections.deque(maxlen=max_entries) for O(1) append with automatic eviction
record_send() creates a TranscriptEntry(timestamp=time.monotonic(), direction="send", text=text)
record_recv() creates a TranscriptEntry(timestamp=time.monotonic(), direction="recv", text=text)
record_gmcp_send() / record_gmcp_recv() serialize the data dict to JSON and prefix with package name
entries property returns a list copy of the deque (newest last)
recent(n) returns the last n entries
format_recent(n, relative_to) returns a formatted string for error messages:
```
[0.00s] RECV: "Character Summary"
[0.01s] RECV: "Create this character?"
[0.50s] SEND: "Y"
[0.52s] RECV: "Character created successfully."
```
If relative_to is None, uses the timestamp of the first entry in the returned set as the base.

2.1.4 Unit Tests — Data Models¶

[ ] Write tests in packages/teltest/tests/test_expect.py:
[ ] test_match_frozen — Verify Match is immutable (assigning to match.text raises FrozenInstanceError)
[ ] test_match_str_representation — Verify __str__ output format
[ ] test_match_fields — Verify all fields are accessible and correctly typed
[ ] test_match_buffer_before_content — Verify buffer_before contains context lines
[ ] test_transcript_entry_frozen — Verify TranscriptEntry is immutable
[ ] test_transcript_entry_directions — Verify all four direction literals are valid
[ ] test_transcript_record_send — Record a send, verify entry fields
[ ] test_transcript_record_recv — Record a recv, verify entry fields
[ ] test_transcript_record_gmcp_send — Record GMCP send, verify package+JSON text format
[ ] test_transcript_record_gmcp_recv — Record GMCP recv, verify package+JSON text format
[ ] test_transcript_ring_buffer_eviction — Create transcript with max_entries=3, record 5 entries, verify only last 3 remain
[ ] test_transcript_recent_returns_n_entries — Record 10 entries, recent(3) returns last 3
[ ] test_transcript_recent_returns_all_if_fewer — Record 2 entries, recent(5) returns both
[ ] test_transcript_format_recent_output — Verify formatted string matches expected pattern with timestamps and direction labels
[ ] test_transcript_format_recent_relative_timestamps — Verify timestamps are relative to base

2.2 Exception Classes¶

Package: teltest | Priority: P0 | Dependencies: 2.1

2.2.1 `ExpectTimeout`¶

[ ] Create ExpectTimeout in packages/teltest/src/teltest/expect.py:

class ExpectTimeout(AssertionError):
    """Raised when expect() times out.

    Provides rich error formatting with full buffer dump,
    transcript excerpt, and diagnostic hints.
    """

    def __init__(
        self,
        pattern: str,
        timeout: float,
        buffer: str,
        transcript: Transcript | None = None,
        hint: str | None = None,
    ) -> None: ...

    @property
    def pattern(self) -> str: ...

    @property
    def timeout(self) -> float: ...

    @property
    def buffer(self) -> str: ...

Extends AssertionError so pytest treats it as a test assertion failure (not an unexpected exception)
__init__ stores pattern, timeout, buffer as instance attributes

__str__ produces the rich error format from the design doc §Error Handling:

teltest.ExpectTimeout: Pattern not found within {timeout}s

  Expected: "{pattern}"

  Received (last 20 lines):
  ─────────────────────────
  |  {line 1}
  |  {line 2}
  |  ...
  ─────────────────────────

  Transcript (last 5 interactions):
    [0.00s] RECV: "..."
    [0.50s] SEND: "..."
    [5.00s] TIMEOUT waiting for "{pattern}"

  Hint: {hint or default hint}

Default hint: "The server may be waiting for additional input. Check if a prompt was expected before this step."
If transcript is None, the "Transcript" section is omitted
Buffer display shows the last 20 lines of the buffer parameter, each prefixed with |
The buffer section uses Unicode box-drawing characters (─) for visual separation

2.2.2 `ConnectionClosed`¶

[ ] Create ConnectionClosed in packages/teltest/src/teltest/expect.py:

class ConnectionClosed(Exception):
    """Raised when the server closes the connection during expect().

    Attributes:
        buffer: Text received before the connection closed.
    """

    def __init__(self, buffer: str, message: str | None = None) -> None: ...

    @property
    def buffer(self) -> str: ...

__init__ stores buffer; sets default message to "Server closed the connection"

__str__ produces:

teltest.ConnectionClosed: Server closed the connection

  Last received text (last 10 lines):
  ─────────────────────────
  |  {line 1}
  |  ...
  ─────────────────────────

  Hint: The server may have crashed or the character was disconnected.
        Check server logs for errors.

Does NOT extend AssertionError — connection closure is an infrastructure failure, not a test assertion

2.2.3 `UnexpectedMatch`¶

[ ] Create UnexpectedMatch in packages/teltest/src/teltest/expect.py:

class UnexpectedMatch(AssertionError):
    """Raised by expect_not() when the forbidden pattern appears.

    Attributes:
        pattern: The pattern that should not have appeared.
        match: The Match object showing where it appeared.
        until: The boundary pattern that was being waited for.
    """

    def __init__(
        self,
        pattern: str,
        match: Match,
        until: str,
    ) -> None: ...

    @property
    def pattern(self) -> str: ...

    @property
    def match(self) -> Match: ...

    @property
    def until(self) -> str: ...

Extends AssertionError — this is a test assertion failure (the negative expectation was violated)

__str__ produces:

teltest.UnexpectedMatch: Forbidden pattern found before boundary

  Forbidden: "{pattern}"
  Boundary:  "{until}"

  Found "{match.text}" on line: "{match.line}"

  Context:
  ─────────────────────────
  |  {match.buffer_before}
  ─────────────────────────

2.2.4 Unit Tests — Exceptions¶

[ ] Write tests in packages/teltest/tests/test_expect.py:
[ ] test_expect_timeout_is_assertion_error — assert issubclass(ExpectTimeout, AssertionError)
[ ] test_expect_timeout_attributes — Verify .pattern, .timeout, .buffer stored correctly
[ ] test_expect_timeout_str_format — Verify str(exc) contains pattern, timeout, buffer lines, hint
[ ] test_expect_timeout_str_with_transcript — Verify transcript section appears when transcript is provided
[ ] test_expect_timeout_str_without_transcript — Verify transcript section is omitted when transcript is None
[ ] test_expect_timeout_str_with_custom_hint — Verify custom hint replaces default
[ ] test_expect_timeout_buffer_truncated_to_20_lines — Provide 50-line buffer, verify only last 20 displayed
[ ] test_connection_closed_is_not_assertion_error — assert not issubclass(ConnectionClosed, AssertionError)
[ ] test_connection_closed_attributes — Verify .buffer stored correctly
[ ] test_connection_closed_str_format — Verify str(exc) contains buffer lines and hint
[ ] test_connection_closed_custom_message — Verify custom message overrides default
[ ] test_unexpected_match_is_assertion_error — assert issubclass(UnexpectedMatch, AssertionError)
[ ] test_unexpected_match_attributes — Verify .pattern, .match, .until stored correctly
[ ] test_unexpected_match_str_format — Verify str(exc) contains forbidden pattern, boundary, matched text, context

2.3 Timeout Scaling¶

Package: teltest | Priority: P0 | Dependencies: none

2.3.1 Timeout Multiplier Helper¶

[ ] Create _resolve_timeout helper function in packages/teltest/src/teltest/expect.py:

_TIMEOUT_MULTIPLIER: float | None = None

def _get_timeout_multiplier() -> float:
    """Read TELTEST_TIMEOUT_MULTIPLIER from environment (cached)."""
    global _TIMEOUT_MULTIPLIER
    if _TIMEOUT_MULTIPLIER is None:
        _TIMEOUT_MULTIPLIER = float(os.environ.get("TELTEST_TIMEOUT_MULTIPLIER", "1.0"))
    return _TIMEOUT_MULTIPLIER

def _resolve_timeout(explicit: float | None, default: float) -> float:
    """Resolve an explicit or default timeout, applying the CI multiplier.

    Args:
        explicit: Timeout passed by the caller (None means use default).
        default: The MUDClient's default timeout.

    Returns:
        Final timeout in seconds, scaled by TELTEST_TIMEOUT_MULTIPLIER.
    """
    base = explicit if explicit is not None else default
    return base * _get_timeout_multiplier()

The multiplier is read once from TELTEST_TIMEOUT_MULTIPLIER and cached in a module global
Default multiplier is 1.0 (no scaling)
CI environments set TELTEST_TIMEOUT_MULTIPLIER=2.0 per the design doc §CI/CD

2.3.2 Unit Tests — Timeout Scaling¶

[ ] Write tests in packages/teltest/tests/test_expect.py:
[ ] test_resolve_timeout_explicit_value — _resolve_timeout(3.0, 5.0) returns 3.0 (no multiplier set)
[ ] test_resolve_timeout_uses_default — _resolve_timeout(None, 5.0) returns 5.0
[ ] test_resolve_timeout_with_multiplier — Set TELTEST_TIMEOUT_MULTIPLIER=2.0 in env, verify _resolve_timeout(3.0, 5.0) returns 6.0
[ ] test_resolve_timeout_default_with_multiplier — Set multiplier to 2.0, verify _resolve_timeout(None, 5.0) returns 10.0
[ ] test_resolve_timeout_multiplier_default_is_1 — Unset env var, verify multiplier is 1.0
Use monkeypatch.setenv / monkeypatch.delenv for environment variable isolation
Reset _TIMEOUT_MULTIPLIER = None in fixture teardown to clear the cache between tests

2.4 Core Expect Engine¶

Package: teltest | Priority: P0 | Dependencies: 2.1, 2.2, 2.3, Phase 1 (MUDClient, OutputBuffer)

The expectation engine is implemented as methods on MUDClient. The internal matching logic lives in packages/teltest/src/teltest/expect.py as free functions, and MUDClient delegates to them.

2.4.1 Internal Matching Helpers¶

[ ] Create _match_pattern function in packages/teltest/src/teltest/expect.py:

def _match_pattern(
    text: str,
    pattern: str | re.Pattern[str],
    start_pos: int = 0,
) -> tuple[int, int, str] | None:
    """Search for pattern in text starting at start_pos.

    Returns:
        (match_start, match_end, matched_text) or None if not found.
    """

For str patterns: use text.find(pattern, start_pos). Return (idx, idx + len(pattern), pattern).
For re.Pattern patterns: use pattern.search(text, pos=start_pos). Return (m.start(), m.end(), m.group(0)).
Returns None if no match found.

[ ] Create _extract_line function in packages/teltest/src/teltest/expect.py:

def _extract_line(text: str, pos: int) -> str:
    """Extract the full line containing the character at pos.

    Scans backward to the previous \\n and forward to the next \\n
    to find the complete line boundaries.
    """

Scans backward from pos to find \n (or start of string)
Scans forward from pos to find \n (or end of string)
Returns the text between those boundaries, stripped of \r\n

[ ] Create _extract_context function in packages/teltest/src/teltest/expect.py:

def _extract_context(text: str, pos: int, context_lines: int = _CONTEXT_LINES) -> str:
    """Extract the last N lines of text before pos for debugging context."""

Splits text[:pos] by \n, takes the last context_lines lines, joins with \n

2.4.2 `expect()` Method¶

[ ] Implement expect() on MUDClient in packages/teltest/src/teltest/client.py:

async def expect(
    self,
    pattern: str | re.Pattern[str],
    *,
    timeout: float | None = None,
    from_start: bool = False,
) -> Match:

Algorithm:
1. Compute effective timeout: effective = _resolve_timeout(timeout, self._timeout)
2. Record start time: start = time.monotonic()
3. Compute deadline: deadline = start + effective
4. Determine search start position:
5. If from_start=True: search_from = 0
6. Else: search_from = self._cursor
7. Scan loop: a. Check for reader error: if self._reader_error is not None, raise ConnectionClosed(buffer=self._buffer.get_text(self._cursor)) b. Check for disconnection: if not self._connected and no reader error, raise ConnectionClosed(buffer=self._buffer.get_text(self._cursor)) c. Get current buffer text: text = self._buffer.get_text(search_from) d. Attempt match: result = _match_pattern(text, pattern) e. If match found:
  - Compute absolute positions: abs_start = search_from + result[0], abs_end = search_from + result[1]
  - If not from_start: advance cursor past match: self._cursor = abs_end
  - Extract line: line = _extract_line(self._buffer.get_all_text(), abs_start)
  - Extract context: context = _extract_context(self._buffer.get_all_text(), abs_start)
  - Compute elapsed: elapsed = time.monotonic() - start
  - Return Match(text=result[2], line=line, pattern=_pattern_str(pattern), elapsed=elapsed, buffer_before=context) f. If no match and deadline exceeded:
  - Raise ExpectTimeout(pattern=_pattern_str(pattern), timeout=effective, buffer=self._buffer.get_text(self._cursor), transcript=self._transcript) g. Wait for new data with remaining timeout: remaining = deadline - time.monotonic()
  - self._new_data.clear()
  - await asyncio.wait_for(self._new_data.wait(), timeout=max(0.01, remaining))
  - On asyncio.TimeoutError: continue loop (will hit deadline check on next iteration)
8. The _new_data event is set by the background reader whenever new data arrives in the buffer

[ ] Add _pattern_str helper:

def _pattern_str(pattern: str | re.Pattern[str]) -> str:
    """Convert pattern to string representation for error messages."""
    if isinstance(pattern, str):
        return pattern
    return pattern.pattern

2.4.3 `expect_prompt()` Method¶

[ ] Implement expect_prompt() on MUDClient in packages/teltest/src/teltest/client.py:

async def expect_prompt(
    self,
    prompt: str | re.Pattern[str] = "> ",
    *,
    timeout: float | None = None,
) -> str:

Multi-signal detection algorithm (per design doc §2):
1. Compute effective timeout and deadline
2. Scan loop: a. Signal 1 — Telnet GA: Check if self._protocol.ga_received flag is set since the last cursor advance. If GA received and buffer has unterminated line matching prompt, match immediately. Clear the GA flag. b. Signal 2 — Prompt regex: If buffer has an unterminated line (no trailing \r\n), match it against prompt (substring for str, search for re.Pattern). If match found, this is a candidate — proceed to quiescence check. c. Signal 3 — Quiescence: If candidate found, wait self._prompt_settle_time (default 0.3s). If no new data arrives during that window, accept the match. d. If no candidate: wait for new data (same as expect()) e. On timeout: raise ExpectTimeout
3. On match: advance cursor past the prompt, record in transcript, return the matched prompt text
Add _prompt_settle_time: float = 0.3 attribute to MUDClient.__init__ (configurable)
Add prompt_settle_time parameter to MUDClient.__init__ constructor

2.4.4 `expect_line()` Method¶

[ ] Implement expect_line() on MUDClient in packages/teltest/src/teltest/client.py:

async def expect_line(
    self,
    text: str,
    *,
    timeout: float | None = None,
) -> str:

Algorithm:
1. Compute effective timeout and deadline
2. Scan loop: a. Get buffer text from cursor: buf = self._buffer.get_text(self._cursor) b. Split into complete lines (lines terminated by \r\n or \n) c. For each complete line, strip trailing whitespace and compare with text.strip() d. If exact match found:
  - Advance cursor past the line (including the newline delimiter)
  - Return the matched line text e. On timeout: raise ExpectTimeout
3. Wait for new data between scan attempts

2.4.5 `expect_sequence()` Method¶

[ ] Implement expect_sequence() on MUDClient in packages/teltest/src/teltest/client.py:

async def expect_sequence(
    self,
    patterns: list[str | re.Pattern[str]],
    *,
    timeout: float | None = None,
    per_step_timeout: float | None = None,
) -> list[Match]:

Timeout semantics (per design doc §2, finding #11):
- If per_step_timeout is provided: each pattern gets per_step_timeout seconds (total time is unbounded, up to len(patterns) * per_step_timeout)
- If only timeout is provided (default): timeout is the total time for all patterns combined. Each individual step gets at most the remaining time.
- If neither is provided: use self._timeout as the total timeout.
Algorithm:
1. Resolve total timeout: total = _resolve_timeout(timeout, self._timeout)
2. Record start time and compute global deadline: deadline = start + total
3. matches: list[Match] = []
4. For each pattern in patterns: a. If per_step_timeout is not None:
  - step_timeout = _resolve_timeout(per_step_timeout, per_step_timeout) (apply multiplier) b. Else:
  - remaining = deadline - time.monotonic()
  - If remaining <= 0: raise ExpectTimeout with info about which step failed (index, pattern)
  - step_timeout = remaining c. Try: match = await self.expect(pattern, timeout=step_timeout) — Note: pass the raw step_timeout value; do NOT let expect() apply the multiplier again. Use an internal _expect_raw() or pass a sentinel to skip double-multiplication. d. Append match to matches
5. Return matches
Important: To avoid double-applying the timeout multiplier, implement an internal _expect_impl() that takes pre-resolved timeouts, and have both expect() and expect_sequence() call it.

2.4.6 `expect_any()` Method¶

[ ] Implement expect_any() on MUDClient in packages/teltest/src/teltest/client.py:

async def expect_any(
    self,
    patterns: list[str | re.Pattern[str]],
    *,
    timeout: float | None = None,
) -> tuple[int, Match]:

Algorithm:
1. Compute effective timeout and deadline
2. Scan loop: a. Get buffer text from cursor: text = self._buffer.get_text(self._cursor) b. For each (index, pattern) in enumerate(patterns):
  - result = _match_pattern(text, pattern)
  - If match found: record match position as (index, abs_start, result) c. If one or more patterns matched:
  - Select the match with the earliest position in the buffer (lowest abs_start). If two patterns match at the same position, prefer the one with the lower index.
  - Advance cursor past the winning match
  - Build and return (winning_index, Match(...)) d. On timeout: raise ExpectTimeout — list all patterns in the error message e. Wait for new data
Design decision: "First match in buffer" (positional) rather than "first match detected" (temporal) ensures deterministic behavior regardless of data arrival timing.

2.4.7 `expect_not()` Method¶

[ ] Implement expect_not() on MUDClient in packages/teltest/src/teltest/client.py:

async def expect_not(
    self,
    pattern: str | re.Pattern[str],
    *,
    until: str | re.Pattern[str],
    timeout: float | None = None,
) -> None:

Algorithm (per design doc §2, finding #4):
1. Compute effective timeout and deadline
2. Record cursor position at entry: entry_cursor = self._cursor
3. Scan loop: a. Get buffer text from cursor: text = self._buffer.get_text(self._cursor) b. Check for forbidden pattern: forbidden = _match_pattern(text, pattern) c. Check for boundary pattern: boundary = _match_pattern(text, until) d. Case 1: Both found — compare positions:
  - If forbidden starts before boundary: raise UnexpectedMatch(pattern=_pattern_str(pattern), match=<Match from forbidden>, until=_pattern_str(until))
  - If boundary starts before or at forbidden: boundary wins — advance cursor past boundary, return None (success) e. Case 2: Only forbidden found (no boundary yet):
  - Raise UnexpectedMatch f. Case 3: Only boundary found (no forbidden):
  - Advance cursor past boundary, return None (success — forbidden pattern never appeared) g. Case 4: Neither found:
  - On timeout: raise ExpectTimeout with pattern set to the until boundary pattern (we timed out waiting for the boundary, not the forbidden pattern)
  - Wait for new data
Important: expect_not() does NOT advance the cursor incrementally during scanning — it preserves the entry cursor until resolution. This ensures the forbidden pattern check covers ALL text between the entry point and the boundary, not just new increments.
Returns None on success (the forbidden pattern was not found before the boundary)

2.4.8 Cursor Management Integration¶

[ ] Ensure MUDClient.send() advances the cursor:

async def send(self, text: str) -> None:
    # ... write to socket ...
    self._cursor = self._buffer.current_length
    self._transcript.record_send(text)

Per design doc §2: "send() advances the cursor to the current buffer end"
This ensures expect() after send() only matches NEW output

[ ] Ensure MUDClient.clear_buffer() resets cursor:

def clear_buffer(self) -> None:
    self._buffer.clear()
    self._cursor = 0

[ ] Add output property to MUDClient:

@property
def output(self) -> str:
    """All received text since connection."""
    return self._buffer.get_all_text()

[ ] Add recent property to MUDClient:

@property
def recent(self) -> str:
    """Text received since last send()/expect() call."""
    return self._buffer.get_text(self._cursor)

[ ] Add transcript property to MUDClient:

@property
def transcript(self) -> list[TranscriptEntry]:
    """Full session transcript with timestamps."""
    return self._transcript.entries

2.4.9 Transcript Integration with Reader¶

[ ] Modify the background reader task in MUDClient (Phase 1 code in client.py) to record received text:

# In the reader loop, after appending clean text to buffer:
if clean_text:
    self._transcript.record_recv(clean_text)

[ ] Initialize self._transcript = Transcript(max_entries=10_000) in MUDClient.__init__

2.5 Public API Exports¶

Package: teltest | Priority: P0 | Dependencies: 2.1–2.4

[ ] Update packages/teltest/src/teltest/__init__.py to export:

from teltest.expect import (
    Match,
    TranscriptEntry,
    Transcript,
    ExpectTimeout,
    ConnectionClosed,
    UnexpectedMatch,
)
from teltest.client import MUDClient

__all__ = [
    "MUDClient",
    "Match",
    "TranscriptEntry",
    "Transcript",
    "ExpectTimeout",
    "ConnectionClosed",
    "UnexpectedMatch",
]

[ ] Verify all public types have Google-style docstrings
[ ] Verify all public method signatures have full type annotations (MyPy strict)

2.6 Unit Tests — Expectation Methods¶

Package: teltest | Priority: P0 | Dependencies: 2.4

All tests use a mock OutputBuffer and mock transport to avoid real network I/O. The test helper FakeMUDClient wraps a MUDClient with an in-process buffer that can be fed data programmatically.

2.6.1 Test Infrastructure¶

[ ] Create test helper in packages/teltest/tests/conftest.py:

class FakeTransport:
    """Mock asyncio transport for testing without network I/O."""

    def __init__(self) -> None:
        self.sent: list[bytes] = []
        self.closed: bool = False

    def write(self, data: bytes) -> None:
        self.sent.append(data)

    def close(self) -> None:
        self.closed = True

    def is_closing(self) -> bool:
        return self.closed

class BufferFeeder:
    """Feeds data into a MUDClient's buffer as if received from the network.

    Usage:
        feeder = BufferFeeder(client)
        feeder.feed("Welcome to the MUD\r\n")
        feeder.feed("Your choice: ")
        match = await client.expect("Welcome")
    """

    def __init__(self, client: MUDClient) -> None: ...

    def feed(self, text: str) -> None:
        """Append text to the client's output buffer and signal new data."""

    def feed_after(self, text: str, delay: float) -> asyncio.Task:
        """Schedule text to be fed after a delay (seconds)."""

    def close_connection(self) -> None:
        """Simulate server closing the connection."""

feed() directly appends to client._buffer, increments buffer length, and sets client._new_data event
feed_after() creates an asyncio.Task that sleeps then calls feed()
close_connection() sets client._connected = False and sets client._new_data event to unblock waiting expects

[ ] Create make_client fixture in packages/teltest/tests/conftest.py:

@pytest.fixture
def make_client() -> Callable[..., tuple[MUDClient, BufferFeeder]]:
    """Factory fixture that creates a MUDClient with a BufferFeeder.

    Returns a (client, feeder) tuple. The client is pre-connected
    (mocked) so expect() can be called immediately.
    """

2.6.2 `expect()` Tests¶

[ ] Write tests in packages/teltest/tests/test_expect.py:

Success cases: - [ ] test_expect_substring_match — Feed "Welcome to the MUD\r\n", expect("Welcome") returns Match with text="Welcome", line="Welcome to the MUD" - [ ] test_expect_regex_match — Feed "Level 42 Warrior\r\n", expect(re.compile(r"Level \d+")) returns Match with text="Level 42" - [ ] test_expect_regex_match_group — Feed "HP: 100/100\r\n", expect(re.compile(r"HP: (\d+)/(\d+)")) returns Match with text="HP: 100/100" - [ ] test_expect_match_elapsed_time — Feed data after 0.1s delay, verify match.elapsed >= 0.1 - [ ] test_expect_match_buffer_before — Feed 5 lines then target, verify match.buffer_before contains preceding lines - [ ] test_expect_match_pattern_string — Verify match.pattern is the string representation of the pattern used - [ ] test_expect_immediate_match — Feed data before calling expect, verify immediate return with elapsed ≈ 0

Cursor behavior: - [ ] test_expect_advances_cursor — Feed "AAA\r\nBBB\r\n", expect("AAA"), then expect("BBB") succeeds (cursor moved past AAA) - [ ] test_expect_does_not_match_before_cursor — Feed "AAA\r\nBBB\r\n", expect("AAA"), then expect("AAA", timeout=0.1) raises ExpectTimeout (cursor is past first AAA) - [ ] test_send_advances_cursor — Feed "AAA\r\n", call send("cmd"), feed "BBB\r\n", expect("AAA", timeout=0.1) raises ExpectTimeout (send moved cursor past AAA) - [ ] test_expect_from_start_ignores_cursor — Feed "AAA\r\nBBB\r\n", expect("AAA") (moves cursor), then expect("AAA", from_start=True) succeeds - [ ] test_expect_from_start_does_not_advance_cursor — Feed "AAA\r\nBBB\r\n", call expect("AAA", from_start=True), then expect("AAA") still works (cursor unchanged by from_start) - [ ] test_clear_buffer_resets_cursor — Feed "AAA\r\n", call clear_buffer(), feed "BBB\r\n", expect("AAA", timeout=0.1) raises ExpectTimeout

Timeout cases: - [ ] test_expect_timeout_raises — Call expect("NEVER", timeout=0.1) with no data, verify ExpectTimeout raised - [ ] test_expect_timeout_exception_attributes — Verify raised exception has correct .pattern, .timeout, .buffer - [ ] test_expect_timeout_exception_has_transcript — Verify raised exception includes transcript entries - [ ] test_expect_timeout_default_timeout — Create client with timeout=0.2, call expect("NEVER") without explicit timeout, verify times out at ~0.2s - [ ] test_expect_timeout_multiplier — Set TELTEST_TIMEOUT_MULTIPLIER=2.0, create client with timeout=0.1, call expect("NEVER"), verify times out at ~0.2s

Partial match / data arrival: - [ ] test_expect_waits_for_complete_data — Feed "Wel" immediately, feed "come\r\n" after 0.1s, expect("Welcome") succeeds after ~0.1s - [ ] test_expect_matches_across_chunks — Feed "Hello " then "World\r\n" in separate chunks, expect("Hello World") succeeds

Connection closed: - [ ] test_expect_connection_closed_raises — Close connection during expect, verify ConnectionClosed raised - [ ] test_expect_connection_closed_has_buffer — Verify ConnectionClosed.buffer contains text received before close - [ ] test_expect_reader_error_raises_connection_closed — Set _reader_error to an exception, call expect(), verify ConnectionClosed raised

2.6.3 `expect_prompt()` Tests¶

[ ] Write tests in packages/teltest/tests/test_expect.py:
[ ] test_expect_prompt_default_pattern — Feed "Room description\r\n> ", expect_prompt() returns "> "
[ ] test_expect_prompt_custom_pattern — Feed "Room\r\nwarrior> ", expect_prompt("warrior> ") returns "warrior> "
[ ] test_expect_prompt_regex_pattern — Feed "Room\r\nmage> ", expect_prompt(re.compile(r"\w+> ")) returns "mage> "
[ ] test_expect_prompt_ga_signal — Set GA received flag, feed "Enter name: ", expect_prompt("Enter name: ") matches immediately without quiescence wait
[ ] test_expect_prompt_quiescence_detection — Feed "Some prompt: " (no \r\n), no GA signal, verify prompt detected after prompt_settle_time elapses
[ ] test_expect_prompt_not_triggered_by_partial_line — Feed "Partial", then feed " line\r\n" before settle time, verify no false prompt detection
[ ] test_expect_prompt_advances_cursor — After prompt match, verify cursor is past the prompt
[ ] test_expect_prompt_timeout — No prompt arrives, verify ExpectTimeout raised
[ ] test_expect_prompt_custom_settle_time — Create client with prompt_settle_time=0.1, verify quiescence detection uses 0.1s

2.6.4 `expect_line()` Tests¶

[ ] Write tests in packages/teltest/tests/test_expect.py:
[ ] test_expect_line_exact_match — Feed "Hello World\r\n", expect_line("Hello World") returns "Hello World"
[ ] test_expect_line_strips_whitespace — Feed "Hello World \r\n", expect_line("Hello World") matches (trailing whitespace stripped)
[ ] test_expect_line_no_partial_match — Feed "Hello World Extra\r\n", expect_line("Hello World", timeout=0.1) raises ExpectTimeout (must be exact)
[ ] test_expect_line_no_substring_match — Feed "Hello World\r\n", expect_line("Hello", timeout=0.1) raises ExpectTimeout (must match full line)
[ ] test_expect_line_skips_non_matching_lines — Feed "Line 1\r\nTarget Line\r\nLine 3\r\n", expect_line("Target Line") matches second line
[ ] test_expect_line_advances_cursor — After match, subsequent expect_line("Target Line", timeout=0.1) raises ExpectTimeout
[ ] test_expect_line_timeout — No matching line arrives, verify ExpectTimeout
[ ] test_expect_line_waits_for_newline — Feed "Hello World" (no \r\n), then feed "\r\n" after delay, verify matches only after newline arrives
[ ] test_expect_line_cursor_behavior — Feed "A\r\nB\r\nC\r\n", expect_line("A"), then expect_line("B") succeeds

2.6.5 `expect_sequence()` Tests¶

[ ] Write tests in packages/teltest/tests/test_expect.py:
[ ] test_expect_sequence_all_match — Feed "Name: \r\nRace: \r\nClass: \r\n", expect_sequence(["Name:", "Race:", "Class:"]) returns list of 3 Match objects
[ ] test_expect_sequence_returns_correct_matches — Verify each Match in the returned list has the correct .text and .pattern
[ ] test_expect_sequence_order_matters — Feed "Class: \r\nName: \r\n", expect_sequence(["Name:", "Class:"]) — only "Name:" is found (after cursor), "Class:" was already consumed or before cursor
[ ] test_expect_sequence_total_timeout — Feed first pattern, delay beyond total timeout for second, verify ExpectTimeout raised
[ ] test_expect_sequence_per_step_timeout — Set per_step_timeout=0.1, feed first pattern immediately, delay second pattern 0.2s, verify ExpectTimeout on second step
[ ] test_expect_sequence_total_timeout_shared — Set timeout=0.5, feed 3 patterns each taking 0.1s, verify all match within 0.5s budget
[ ] test_expect_sequence_cursor_advances_through — After sequence completes, cursor is past all matched patterns
[ ] test_expect_sequence_empty_patterns_returns_empty — expect_sequence([]) returns [] immediately
[ ] test_expect_sequence_partial_failure_reports_step — Second of three patterns times out; verify error message indicates which step failed
[ ] test_expect_sequence_with_regex_patterns — Mix of string and regex patterns all match correctly
[ ] test_expect_sequence_no_double_timeout_multiplier — Set TELTEST_TIMEOUT_MULTIPLIER=2.0 and timeout=1.0, verify total timeout is 2.0s (not 4.0s from double multiplication)

2.6.6 `expect_any()` Tests¶

[ ] Write tests in packages/teltest/tests/test_expect.py:
[ ] test_expect_any_first_pattern_matches — Feed "Yes\r\n", expect_any(["Yes", "No"]) returns (0, Match(text="Yes"))
[ ] test_expect_any_second_pattern_matches — Feed "No\r\n", expect_any(["Yes", "No"]) returns (1, Match(text="No"))
[ ] test_expect_any_returns_earliest_position — Feed "No then Yes\r\n", expect_any(["Yes", "No"]) returns (1, Match(text="No")) because "No" appears first positionally
[ ] test_expect_any_same_position_prefers_lower_index — Feed "AB\r\n" where patterns ["AB", "A"] both match at pos 0, returns (0, Match(text="AB"))
[ ] test_expect_any_with_regex — Feed "HP: 42\r\n", expect_any([re.compile(r"HP: \d+"), "MP:"]) returns (0, Match(text="HP: 42"))
[ ] test_expect_any_advances_cursor — After match, verify cursor is past the matched text
[ ] test_expect_any_timeout — None of the patterns appear, verify ExpectTimeout with all patterns listed
[ ] test_expect_any_waits_for_data — Feed matching data after 0.1s delay, verify it waits and returns
[ ] test_expect_any_single_pattern — expect_any(["Only"]) with matching data works like expect("Only")

2.6.7 `expect_not()` Tests¶

[ ] Write tests in packages/teltest/tests/test_expect.py:

Success cases (forbidden pattern NOT found): - [ ] test_expect_not_success — Feed "Room description\r\n> ", expect_not("Error", until="> ") returns None - [ ] test_expect_not_boundary_found_first — Feed "Good stuff\r\n> ", verify returns None - [ ] test_expect_not_advances_cursor_past_boundary — After success, cursor is past the until match

Failure cases (forbidden pattern found): - [ ] test_expect_not_forbidden_found_before_boundary — Feed "Error: bad\r\n> ", expect_not("Error", until="> ") raises UnexpectedMatch - [ ] test_expect_not_unexpected_match_attributes — Verify raised exception has correct .pattern, .match, .until - [ ] test_expect_not_forbidden_found_no_boundary — Feed "Error occurred\r\n" only, expect_not("Error", until="> ") raises UnexpectedMatch

Timeout cases: - [ ] test_expect_not_timeout_waiting_for_boundary — Feed "Stuff\r\n" (no boundary, no forbidden), expect_not("Error", until="> ", timeout=0.1) raises ExpectTimeout for the boundary pattern - [ ] test_expect_not_timeout_message_references_boundary — Verify the ExpectTimeout.pattern is the until pattern, not the forbidden pattern

Edge cases: - [ ] test_expect_not_with_regex_patterns — Both pattern and until are re.Pattern, verify correct behavior - [ ] test_expect_not_forbidden_and_boundary_overlap — Feed "> Error >", where forbidden="Error" and until=">", verify boundary at position 0 wins (boundary found first) - [ ] test_expect_not_cursor_preserved_during_scan — Verify the cursor does not advance incrementally while scanning; it only advances to the boundary match position on success

2.6.8 `output` / `recent` / `clear_buffer()` / `transcript` Tests¶

[ ] Write tests in packages/teltest/tests/test_expect.py:
[ ] test_output_returns_all_text — Feed multiple chunks, verify client.output contains all text
[ ] test_recent_returns_text_since_cursor — Feed text, call send(), feed more, verify client.recent only shows post-send text
[ ] test_recent_after_expect — Feed "A\r\nB\r\n", expect("A"), verify client.recent is "B\r\n" (text after match)
[ ] test_clear_buffer_empties_output — Feed text, clear_buffer(), verify client.output == ""
[ ] test_clear_buffer_resets_recent — Feed text, clear_buffer(), verify client.recent == ""
[ ] test_transcript_records_send_and_recv — Send a command, feed response, verify transcript has both entries in order
[ ] test_transcript_records_correct_timestamps — Verify transcript timestamps are monotonically increasing
[ ] test_transcript_gmcp_entries — Record GMCP send/recv, verify entries appear with correct direction

2.7 Integration Tests¶

Package: teltest | Priority: P0 | Dependencies: 2.1–2.6

2.7.1 End-to-End Expect Flow Tests¶

These tests verify the expect methods work together in realistic multi-step scenarios using the FakeMUDClient infrastructure.

[ ] Write tests in packages/teltest/tests/test_expect_integration.py:
[ ] test_login_flow_simulation — Simulate a login flow:
1. Feed "Welcome to TestMUD\r\nPlease select: [L]ogin [R]egister\r\n> "
2. expect("Welcome")
3. expect_prompt("> ")
4. send("L")
5. Feed "Username: "
6. expect_prompt("Username: ")
7. send("testuser")
8. Feed "Password: "
9. expect_prompt("Password: ")
10. send("testpass")
11. Feed "Login successful!\r\nEntering world...\r\n> "
12. expect("Login successful")
13. expect_prompt("> ") Verify all matches succeed and transcript has all entries in order.
[ ] test_expect_sequence_then_any — Feed a character creation flow:
1. Feed "Name: \r\nRace: \r\nClass: \r\n"
2. expect_sequence(["Name:", "Race:", "Class:"])
3. Feed "Confirm? [Y/N]\r\n"
4. send("Y")
5. Feed "Character created!\r\n" or "Creation failed!\r\n"
6. expect_any(["Character created", "Creation failed"]) → verify returns (0, ...)
[ ] test_expect_not_in_gameplay — Verify no error during command execution:
1. send("look")
2. Feed "A sunny meadow\r\nExits: north south\r\n> "
3. expect_not("Error", until="> ") succeeds
4. expect("sunny meadow", from_start=True) succeeds (verify from_start works after expect_not)
[ ] test_cursor_isolation_across_commands — Multi-command sequence:
1. send("command1")
2. Feed "Response 1\r\n> "
3. expect("Response 1")
4. send("command2")
5. Feed "Response 2\r\n> "
6. expect("Response 2") succeeds
7. expect("Response 1", timeout=0.1) raises ExpectTimeout (cursor past it)
[ ] test_timeout_error_diagnostic_quality — Trigger a timeout and verify the error message contains:
1. The pattern searched for
2. The timeout value
3. The last 20 lines of received output
4. The transcript of recent interactions
5. A diagnostic hint
[ ] test_connection_closed_mid_expect — Start an expect, close connection from feeder, verify ConnectionClosed raised with buffer context
[ ] test_transcript_completeness — Execute a multi-step flow, verify client.transcript contains every send and recv in chronological order with monotonically increasing timestamps

File Summary¶

File	Purpose	New/Modified
`packages/teltest/src/teltest/expect.py`	`Match`, `TranscriptEntry`, `Transcript`, `ExpectTimeout`, `ConnectionClosed`, `UnexpectedMatch`, `_match_pattern`, `_extract_line`, `_extract_context`, `_resolve_timeout`, `_pattern_str`	New
`packages/teltest/src/teltest/client.py`	`expect()`, `expect_prompt()`, `expect_line()`, `expect_sequence()`, `expect_any()`, `expect_not()`, `output`, `recent`, `clear_buffer()`, `transcript`, cursor management	Modified
`packages/teltest/src/teltest/__init__.py`	Public API exports	Modified
`packages/teltest/tests/conftest.py`	`FakeTransport`, `BufferFeeder`, `make_client` fixture	New/Modified
`packages/teltest/tests/test_expect.py`	All unit tests for data models, exceptions, timeout scaling, and expect methods	New
`packages/teltest/tests/test_expect_integration.py`	Integration tests for multi-step expect flows	New

Test Summary¶

Section	Test Count	File
2.1 Data Models	15	`test_expect.py`
2.2 Exceptions	14	`test_expect.py`
2.3 Timeout Scaling	5	`test_expect.py`
2.6.2 `expect()`	23	`test_expect.py`
2.6.3 `expect_prompt()`	9	`test_expect.py`
2.6.4 `expect_line()`	9	`test_expect.py`
2.6.5 `expect_sequence()`	11	`test_expect.py`
2.6.6 `expect_any()`	9	`test_expect.py`
2.6.7 `expect_not()`	11	`test_expect.py`
2.6.8 Properties/Buffer	8	`test_expect.py`
2.7 Integration	7	`test_expect_integration.py`
Total	121

Verification Checklist¶

[ ] uv run pytest packages/teltest/tests/test_expect.py -v — all unit tests pass
[ ] uv run pytest packages/teltest/tests/test_expect_integration.py -v — all integration tests pass
[ ] uv run mypy packages/teltest/src/teltest/expect.py --strict — no type errors
[ ] uv run mypy packages/teltest/src/teltest/client.py --strict — no type errors
[ ] uv run ruff check packages/teltest/ — no lint errors
[ ] All public classes/methods have Google-style docstrings
[ ] ExpectTimeout error message matches the format in design doc §Error Handling
[ ] TELTEST_TIMEOUT_MULTIPLIER=2.0 correctly doubles all timeouts
[ ] No test uses asyncio.sleep for synchronization (use feed_after or events)
[ ] All expect methods respect cursor semantics documented in design doc §2

Estimated Timeline¶

Week	Tasks	Output
1	2.1 (Data Models), 2.2 (Exceptions), 2.3 (Timeout Scaling)	`Match`, `TranscriptEntry`, `Transcript`, all 3 exceptions, timeout helpers, 33 unit tests
2	2.4.1–2.4.4 (`expect`, `expect_prompt`, `expect_line`, helpers)	Core expect loop, prompt detection, exact-line matching, 36 unit tests
3	2.4.5–2.4.8 (`expect_sequence`, `expect_any`, `expect_not`, cursor/transcript)	All remaining expect methods, cursor integration, 39 unit tests
4	2.5 (Exports), 2.6.8 (Property tests), 2.7 (Integration)	Public API, 15 integration + property tests, verification checklist

Expectation Engine — Implementation Plan¶

Summary¶

2.1 Data Models¶

2.1.1 Match Dataclass¶

2.1.2 TranscriptEntry Dataclass¶

2.1.3 Transcript Recorder¶