584@contextmanager
585def terminal_session(
586    command: Command,
587    rows: Rows = None,
588    cols: Cols = None,
589    logger: Logger = None,
590    extra_subscribes: ExtraSubscribes = None,
591) -> Iterator[HtWrapper]:
592    """
593    The terminal_session context manager is a wrapper around `run` which ensures that the underlying process
594    gets cleaned up:
595
596    ```python
597    with terminal_session("some command") as proc:
598        # interact with the running command here
599        assert proc.exit_code is not None
600        s = proc.snapshot()
601
602    # htty terminates your command on context exit
603    assert proc.exit_code is None
604    assert "hello world" in proc.snapshot()
605    ```
606
607    Its usage is otherwise the same as `run`.
608    Like `run` it returns a `HtWrapper`.
609
610    """
611
612    proc = run(
613        command,
614        rows=rows,
615        cols=cols,
616        no_exit=True,
617        logger=logger,
618        extra_subscribes=extra_subscribes,
619    )
620    try:
621        yield proc
622    finally:
623        try:
624            if proc.cmd.pid:
625                proc.cmd.terminate()
626                proc.cmd.wait(timeout=DEFAULT_SUBPROCESS_WAIT_TIMEOUT)
627        except Exception:
628            try:
629                if proc.cmd.pid:
630                    proc.cmd.kill()
631            except Exception:
632                pass
633
634        try:
635            proc.ht.terminate()
636            proc.ht.wait(timeout=DEFAULT_SUBPROCESS_WAIT_TIMEOUT)
637        except Exception:
638            with suppress(Exception):
639                proc.ht.kill()

642def run(
643    command: Command,
644    rows: Rows = None,
645    cols: Cols = None,
646    no_exit: NoExit = True,
647    logger: Logger = None,
648    extra_subscribes: ExtraSubscribes = None,
649) -> HtWrapper:
650    """
651    As a user of the htty python library, your code will run in the python process at the root of this
652    process tree:
653
654        python '{/path/to/your/code.py}'
655        └── ht
656            └── sh -c '{modified command}'
657
658    So if you're using htty to wrap vim, the process tree is:
659
660    ```
661    python
662    └── ht
663        └── sh
664            └── vim
665    ```
666
667    This function invokes `ht` as a subprocess such that you end up with a process tree like the one shown
668    above. It returns an `HtWrapper` object which can be used to interact with ht and its child process.
669
670    It's up to you to clean up this process when you're done:
671
672    ```python
673    proc = run("some command")
674    # do stuff
675    proc.exit()
676    ```
677
678    If you'd rather not risk having a bunch of `ht` processes lying around and wasting CPU cycles,
679    consider using the `terminal_session` instead.
680
681    For reasons that are documented in
682    [htty-core](https://matrixmanatyrservice.github.io/htty/htty-core/htty_core.html#HtEvent.COMMAND_COMPLETED), the
683    command that ht runs is not:
684
685        sh -c '{command}'
686
687    Instead it's something like this:
688
689        sh -c '{command} ; exit_code=$? ; /path/to/ht wait-exit /path/to/tmp/ht_fifo_5432 ; exit $exit_code'
690
691    Because of this, it's possible to come up with command strings that cause sh to behave in problematic ways (for
692    example: `'`). For now the mitigation for this is: "don't do that." (If you'd like me to prioritize changing this
693    please leave a comment in https://github.com/MatrixManAtYrService/htty/issues/2)
694    """
695    # Use provided logger or fall back to default
696    process_logger = logger or default_logger
697
698    # Create a queue for events
699    event_queue: queue.Queue[dict[str, Any]] = queue.Queue()
700
701    # Build the ht subscription list
702    base_subscribes = [
703        HtEvent.INIT,
704        HtEvent.SNAPSHOT,
705        HtEvent.OUTPUT,
706        HtEvent.RESIZE,
707        HtEvent.PID,
708        HtEvent.EXIT_CODE,
709        HtEvent.COMMAND_COMPLETED,
710    ]
711    if extra_subscribes:
712        # Convert string subscribes to HtEvent enum values
713        for sub in extra_subscribes:
714            try:
715                base_subscribes.append(HtEvent(sub))
716            except ValueError:
717                process_logger.warning(f"Unknown subscription event: {sub}")
718
719    # Convert command to string if it's a list, properly escaping shell arguments
720    command_str = command if isinstance(command, str) else " ".join(shlex.quote(arg) for arg in command)
721
722    # Create HtArgs and use htty_core.run()
723    ht_args = HtArgs(
724        command=command_str,  # Use the already-formatted command string
725        subscribes=base_subscribes,
726        rows=rows,
727        cols=cols,
728    )
729
730    # Log the exact command that would be run
731    cmd_args = ht_args.get_command()
732    process_logger.debug(f"Launching command: {' '.join(cmd_args)}")
733
734    ht_proc = htty_core_run(ht_args)
735
736    process_logger.debug(f"ht started: PID {ht_proc.pid}")
737
738    # Create a reader thread to capture ht output
739    def reader_thread(
740        ht_proc: subprocess.Popen[str],
741        queue_obj: queue.Queue[dict[str, Any]],
742        ht_process: HtWrapper,
743        thread_logger: logging.Logger,
744    ) -> None:
745        thread_logger.debug(f"Reader thread started for ht process {ht_proc.pid}")
746
747        while True:
748            if ht_proc.stdout is None:
749                thread_logger.warning(f"ht process {ht_proc.pid} stdout is None, exiting reader thread")
750                break
751
752            line = ht_proc.stdout.readline()
753            if not line:
754                thread_logger.debug(f"ht process {ht_proc.pid} stdout closed, exiting reader thread")
755                break
756
757            line = line.strip()
758            if not line:
759                continue
760
761            try:
762                event = json.loads(line)
763                thread_logger.debug(f"ht event: {event}")
764                queue_obj.put(event)
765
766                if event["type"] == "output":
767                    ht_process.add_output_event(event)
768                elif event["type"] == "exitCode":
769                    thread_logger.debug(
770                        f"ht process {ht_proc.pid} subprocess exited with code: {event.get('data', {}).get('exitCode')}"
771                    )
772                    ht_process.set_subprocess_exited(True)
773                    exit_code = event.get("data", {}).get("exitCode")
774                    if exit_code is not None:
775                        ht_process.cmd.exit_code = exit_code
776                elif event["type"] == "pid":
777                    thread_logger.debug(f"ht process {ht_proc.pid} subprocess PID: {event.get('data', {}).get('pid')}")
778                    pid = event.get("data", {}).get("pid")
779                    if pid is not None:
780                        ht_process.cmd.pid = pid
781                elif event["type"] == "commandCompleted":
782                    # Command has completed - this is the reliable signal that subprocess finished
783                    ht_process.set_subprocess_completed(True)
784                elif event["type"] == "debug":
785                    thread_logger.debug(f"ht process {ht_proc.pid} debug: {event.get('data', {})}")
786                    # Note: We no longer rely on debug events for subprocess_completed
787                    # The commandCompleted event (above) is the reliable source
788            except json.JSONDecodeError as e:
789                # Only log raw stdout when we can't parse it as JSON - this indicates an unexpected message
790                thread_logger.warning(f"ht process {ht_proc.pid} non-JSON stdout: {line} (error: {e})")
791                pass
792
793        thread_logger.debug(f"Reader thread exiting for ht process {ht_proc.pid}")
794
795    # Create an HtWrapper instance
796    process = HtWrapper(
797        ht_proc,
798        event_queue,
799        command=command_str,
800        rows=rows,
801        cols=cols,
802        no_exit=no_exit,
803        logger=process_logger,
804    )
805
806    # Start the reader thread for stdout
807    stdout_thread = threading.Thread(
808        target=reader_thread,
809        args=(ht_proc, event_queue, process, process_logger),
810        daemon=True,
811    )
812    stdout_thread.start()
813
814    # Start a stderr reader thread
815    def stderr_reader_thread(ht_proc: subprocess.Popen[str], thread_logger: logging.Logger) -> None:
816        thread_logger.debug(f"Stderr reader thread started for ht process {ht_proc.pid}")
817
818        while True:
819            if ht_proc.stderr is None:
820                thread_logger.warning(f"ht process {ht_proc.pid} stderr is None, exiting stderr reader thread")
821                break
822
823            line = ht_proc.stderr.readline()
824            if not line:
825                thread_logger.debug(f"ht process {ht_proc.pid} stderr closed, exiting stderr reader thread")
826                break
827
828            line = line.strip()
829            if line:
830                thread_logger.debug(f"ht stderr: {line}")
831
832        thread_logger.debug(f"Stderr reader thread exiting for ht process {ht_proc.pid}")
833
834    stderr_thread = threading.Thread(target=stderr_reader_thread, args=(ht_proc, process_logger), daemon=True)
835    stderr_thread.start()
836
837    # Wait briefly for the process to initialize and get PID
838    start_time = time.time()
839    while time.time() - start_time < 2:
840        try:
841            event = event_queue.get(block=True, timeout=0.5)
842            if event["type"] == "pid":
843                pid = event["data"]["pid"]
844                process.cmd.pid = pid
845                break
846        except queue.Empty:
847            continue
848
849    time.sleep(DEFAULT_SLEEP_AFTER_KEYS)
850    return process

 80class HtWrapper:
 81    """
 82    A wrapper around a process started with the 'ht' tool that provides
 83    methods for interacting with the process and capturing its output.
 84    """
 85
 86    ht: ProcessController
 87    """
 88    Helpers for interacting with the `ht` process (a child process of the python
 89    that called `htty.run` or `htty.terminal_session`)
 90    """
 91
 92    cmd: ProcessController
 93    """
 94    Helpers for interacting with the shell that wraps the caller's command (`ht`'s child process)
 95    """
 96
 97    def __init__(
 98        self,
 99        ht_proc: "subprocess.Popen[str]",
100        event_queue: queue.Queue[dict[str, Any]],
101        command: Optional[str] = None,
102        pid: Optional[int] = None,
103        rows: Optional[int] = None,
104        cols: Optional[int] = None,
105        no_exit: bool = False,
106        logger: Optional[logging.Logger] = None,
107    ) -> None:
108        """
109        @private
110        Users are not expect to create these directly.
111        They should use `with terminal_session(...)` or `run(...)`
112        """
113        self._ht_proc = ht_proc  # The ht process itself
114        self._cmd_process = CmdProcess(pid)
115        self._event_queue = event_queue
116        self._command = command
117        self._output_events: list[dict[str, Any]] = []
118        self._unknown_events: list[dict[str, Any]] = []
119        self._latest_snapshot: Optional[str] = None
120        self._start_time = time.time()
121        self._exit_code: Optional[int] = None
122        self._rows = rows
123        self._cols = cols
124        self._no_exit = no_exit
125        self._subprocess_exited = False
126        self._subprocess_completed = False  # Set earlier when command completion is detected
127
128        # Use provided logger or fall back to default
129        self._logger = logger or default_logger
130        self._logger.debug(f"HTProcess created: ht_proc.pid={ht_proc.pid}, command={command}")
131
132        # Create the public interface objects
133        self.ht: ProcessController = HtProcess(ht_proc, self)
134        self.cmd: ProcessController = self._cmd_process
135
136    def __del__(self):
137        """Destructor to warn about uncleaned processes."""
138        if hasattr(self, "_ht_proc") and self._ht_proc and self._ht_proc.poll() is None:
139            self._logger.warning(
140                f"HTProcess being garbage collected with running ht process (PID: {self._ht_proc.pid}). "
141                f"This may cause resource leaks!"
142            )
143            # Try emergency cleanup
144            with suppress(Exception):
145                self._ht_proc.terminate()
146
147    def get_output(self) -> list[dict[str, Any]]:
148        """
149        Return list of [output](./htty-core/htty_core.html#HtEvent.OUTPUT) events."""
150        return [event for event in self._output_events if event.get("type") == "output"]
151
152    def add_output_event(self, event: dict[str, Any]) -> None:
153        """
154        @private
155        Add an output event (for internal use by reader thread).
156        """
157        self._output_events.append(event)
158
159    def set_subprocess_exited(self, exited: bool) -> None:
160        """
161        @private
162        Set subprocess exited flag (for internal use by reader thread).
163        """
164        self._subprocess_exited = exited
165
166    def set_subprocess_completed(self, completed: bool) -> None:
167        """
168        @private
169        Set subprocess completed flag (for internal use by reader thread).
170        """
171        self._subprocess_completed = completed
172        self._cmd_process.set_completed(completed)
173
174    def send_keys(self, keys: Union[KeyInput, list[KeyInput]]) -> None:
175        """
176        Send keys to the terminal.  Accepts strings, `Press` objects, and lists of strings or `Press` objects.
177        For keys that you can `Press`, see
178        [keys.py](https://github.com/MatrixManAtYrService/htty/blob/main/htty/src/htty/keys.py).
179
180        ```python
181        from htty import Press, terminal_session
182
183        with (
184            terminal_session("sh -i", rows=4, cols=40, logger=test_logger) as sh,
185        ):
186            sh.send_keys("echo foo && sleep 999")
187            sh.send_keys(Press.ENTER)
188            sh.expect("^foo")
189            sh.send_keys(Press.CTRL_Z)
190            sh.expect("Stopped")
191            sh.send_keys(["clear", Press.ENTER])
192        ```
193
194        These are sent to `ht` as events that look like this
195
196        ```json
197        {"type": "sendKeys", "keys": ["echo foo && sleep 999", "Enter"]}
198        ```
199
200        Notice that Press.ENTER is still sent as "Enter" under the hood.
201        `ht` checks to see if it corresponds with a known key and sends it letter-at-a-time if not.
202
203        Because of this, you might run into suprises if you want to type individual characters which happen to spell out
204        a known key such as "Enter" or "Backspace".
205
206        Work around this by breaking up the key names like so:
207
208        ```python
209        sh.send_keys(["Ente", "r"])
210        ```
211
212        If this behavior is problematic for you, consider submitting an issue.
213        """
214        key_strings = keys_to_strings(keys)
215        message = json.dumps({"type": "sendKeys", "keys": key_strings})
216
217        self._logger.debug(f"Sending keys: {message}")
218
219        if self._ht_proc.stdin is not None:
220            try:
221                self._ht_proc.stdin.write(message + "\n")
222                self._ht_proc.stdin.flush()
223                self._logger.debug("Keys sent successfully")
224            except (BrokenPipeError, OSError) as e:
225                self._logger.error(f"Failed to send keys: {e}")
226                self._logger.error(f"ht process poll result: {self._ht_proc.poll()}")
227                raise
228        else:
229            self._logger.error("ht process stdin is None")
230
231        time.sleep(DEFAULT_SLEEP_AFTER_KEYS)
232
233    def snapshot(self, timeout: float = DEFAULT_SNAPSHOT_TIMEOUT) -> SnapshotResult:
234        """
235        Take a snapshot of the terminal output.
236        """
237        if self._ht_proc.poll() is not None:
238            raise RuntimeError(f"ht process has exited with code {self._ht_proc.returncode}")
239
240        message = json.dumps({"type": "takeSnapshot"})
241        self._logger.debug(f"Taking snapshot: {message}")
242
243        try:
244            if self._ht_proc.stdin is not None:
245                self._ht_proc.stdin.write(message + "\n")
246                self._ht_proc.stdin.flush()
247                self._logger.debug("Snapshot request sent successfully")
248            else:
249                raise RuntimeError("ht process stdin is not available")
250        except BrokenPipeError as e:
251            self._logger.error(f"Failed to send snapshot request: {e}")
252            self._logger.error(f"ht process poll result: {self._ht_proc.poll()}")
253            raise RuntimeError(
254                f"Cannot communicate with ht process (broken pipe). "
255                f"Process may have exited. Poll result: {self._ht_proc.poll()}"
256            ) from e
257
258        time.sleep(DEFAULT_SLEEP_AFTER_KEYS)
259
260        # Use tenacity to retry getting the snapshot
261        return self._wait_for_snapshot(timeout)
262
263    def _wait_for_snapshot(self, timeout: float) -> SnapshotResult:
264        """
265        Wait for snapshot response from the event queue with timeout and retries.
266        """
267
268        @retry(
269            stop=stop_after_delay(timeout),
270            wait=wait_fixed(SNAPSHOT_RETRY_TIMEOUT),
271            retry=retry_if_exception_type(SnapshotNotReady),
272        )
273        def _get_snapshot() -> SnapshotResult:
274            try:
275                event = self._event_queue.get(block=True, timeout=SNAPSHOT_RETRY_TIMEOUT)
276            except queue.Empty as e:
277                raise SnapshotNotReady("No events available in queue") from e
278
279            if event["type"] == "snapshot":
280                data = event["data"]
281                snapshot_text = data["text"]
282                raw_seq = data["seq"]
283
284                # Convert to HTML with ANSI color support
285                html = simple_ansi_to_html(raw_seq)
286
287                return SnapshotResult(
288                    text=snapshot_text,
289                    html=html,
290                    raw_seq=raw_seq,
291                )
292            elif event["type"] == "output":
293                self._output_events.append(event)
294            elif event["type"] == "resize":
295                data = event.get("data", {})
296                if "rows" in data:
297                    self._rows = data["rows"]
298                if "cols" in data:
299                    self._cols = data["cols"]
300            elif event["type"] == "init":
301                pass
302            else:
303                # Put non-snapshot events back in queue for reader thread to handle
304                self._event_queue.put(event)
305
306            # If we get here, we didn't find a snapshot, so retry
307            raise SnapshotNotReady(f"Received {event['type']} event, waiting for snapshot")
308
309        try:
310            return _get_snapshot()
311        except Exception as e:
312            # Handle both direct SnapshotNotReady and tenacity RetryError
313            from tenacity import RetryError
314
315            if isinstance(e, (SnapshotNotReady, RetryError)):
316                raise RuntimeError(
317                    f"Failed to receive snapshot event within {timeout} seconds. "
318                    f"ht process may have exited or stopped responding."
319                ) from e
320            raise
321
322    def exit(self, timeout: float = DEFAULT_EXIT_TIMEOUT) -> int:
323        """
324        Exit the ht process, ensuring clean shutdown.
325
326        Uses different strategies based on subprocess state:
327        - If subprocess already exited (exitCode event received): graceful shutdown via exit command
328        - If subprocess still running: forced termination with SIGTERM then SIGKILL
329        """
330        self._logger.debug(f"Exiting HTProcess: ht_proc.pid={self._ht_proc.pid}")
331
332        # Check if we've already received the exitCode event
333        if self._subprocess_exited:
334            self._logger.debug("Subprocess already exited (exitCode event received), attempting graceful shutdown")
335            return self._graceful_exit(timeout)
336        else:
337            self._logger.debug("Subprocess has not exited yet, checking current state")
338
339            # Give a brief moment for any pending exitCode event to arrive
340            brief_wait_start = time.time()
341            while time.time() - brief_wait_start < 0.5:  # Wait up to 500ms
342                if self._subprocess_exited:
343                    self._logger.debug("Subprocess exited during brief wait, attempting graceful shutdown")
344                    return self._graceful_exit(timeout)
345                time.sleep(0.01)
346
347            self._logger.debug("Subprocess still running after brief wait, using forced termination")
348            return self._forced_exit(timeout)
349
350    def _graceful_exit(self, timeout: float) -> int:
351        """
352        Graceful exit: subprocess has completed, so we can send exit command to ht process.
353        """
354        # Send exit command to ht process
355        message = json.dumps({"type": "exit"})
356        self._logger.debug(f"Sending exit command to ht process {self._ht_proc.pid}: {message}")
357
358        try:
359            if self._ht_proc.stdin is not None:
360                self._ht_proc.stdin.write(message + "\n")
361                self._ht_proc.stdin.flush()
362                self._logger.debug(f"Exit command sent successfully to ht process {self._ht_proc.pid}")
363                self._ht_proc.stdin.close()  # Close stdin after sending exit command
364                self._logger.debug(f"Closed stdin for ht process {self._ht_proc.pid}")
365            else:
366                self._logger.debug(f"ht process {self._ht_proc.pid} stdin is None, cannot send exit command")
367        except (BrokenPipeError, OSError) as e:
368            self._logger.debug(
369                f"Failed to send exit command to ht process {self._ht_proc.pid}: {e} (process may have already exited)"
370            )
371            pass
372
373        # Wait for the ht process to finish gracefully
374        start_time = time.time()
375        while self._ht_proc.poll() is None:
376            if time.time() - start_time > timeout:
377                # Graceful exit timed out, fall back to forced termination
378                self._logger.warning(
379                    f"ht process {self._ht_proc.pid} did not exit gracefully within timeout, "
380                    f"falling back to forced termination"
381                )
382                return self._forced_exit(timeout)
383            time.sleep(DEFAULT_SLEEP_AFTER_KEYS)
384
385        self._exit_code = self._ht_proc.returncode
386        if self._exit_code is None:
387            raise RuntimeError("Failed to determine ht process exit code")
388
389        self._logger.debug(f"HTProcess exited gracefully: exit_code={self._exit_code}")
390        return self._exit_code
391
392    def _forced_exit(self, timeout: float) -> int:
393        """
394        Forced exit: subprocess may still be running, so we need to terminate everything forcefully.
395        """
396        # Step 1: Ensure subprocess is terminated first if needed
397        if self._cmd_process.pid and not self._subprocess_exited:
398            self._logger.debug(f"Terminating subprocess: pid={self._cmd_process.pid}")
399            try:
400                os.kill(self._cmd_process.pid, 0)
401                self._cmd_process.terminate()
402                try:
403                    self._cmd_process.wait(timeout=DEFAULT_SUBPROCESS_WAIT_TIMEOUT)
404                    self._logger.debug(f"Subprocess {self._cmd_process.pid} terminated successfully")
405                except Exception:
406                    self._logger.warning(f"Subprocess {self._cmd_process.pid} did not terminate gracefully, killing")
407                    with suppress(Exception):
408                        self._cmd_process.kill()
409            except OSError:
410                self._logger.debug(f"Subprocess {self._cmd_process.pid} already exited")
411                pass  # Process already exited
412
413        # Step 2: Force terminate the ht process with SIGTERM, then SIGKILL if needed
414        self._logger.debug(f"Force terminating ht process {self._ht_proc.pid}")
415
416        # Try SIGTERM first
417        try:
418            self._ht_proc.terminate()
419            self._logger.debug(f"Sent SIGTERM to ht process {self._ht_proc.pid}")
420        except Exception as e:
421            self._logger.debug(f"Failed to send SIGTERM to ht process {self._ht_proc.pid}: {e}")
422
423        # Wait for termination
424        start_time = time.time()
425        while self._ht_proc.poll() is None:
426            if time.time() - start_time > timeout:
427                # SIGTERM timeout, try SIGKILL
428                self._logger.warning(
429                    f"ht process {self._ht_proc.pid} did not terminate with SIGTERM within timeout, sending SIGKILL"
430                )
431                try:
432                    self._ht_proc.kill()
433                    self._logger.debug(f"Sent SIGKILL to ht process {self._ht_proc.pid}")
434                except Exception as e:
435                    self._logger.debug(f"Failed to send SIGKILL to ht process {self._ht_proc.pid}: {e}")
436
437                # Wait for SIGKILL to take effect
438                kill_start_time = time.time()
439                while self._ht_proc.poll() is None:
440                    if time.time() - kill_start_time > timeout:
441                        self._logger.error(f"ht process {self._ht_proc.pid} did not respond to SIGKILL within timeout")
442                        break
443                    time.sleep(DEFAULT_SLEEP_AFTER_KEYS)
444                break
445            time.sleep(DEFAULT_SLEEP_AFTER_KEYS)
446
447        self._exit_code = self._ht_proc.returncode
448        if self._exit_code is None:
449            raise RuntimeError("Failed to determine ht process exit code")
450
451        self._logger.debug(f"HTProcess exited via forced termination: exit_code={self._exit_code}")
452        return self._exit_code
453
454    def expect(self, pattern: str, timeout: float = DEFAULT_EXPECT_TIMEOUT) -> None:
455        """
456        Wait for a regex pattern to appear in the terminal output.
457
458        This method efficiently waits for output by monitoring the output events from
459        the ht process rather than polling with snapshots. It checks both the current
460        terminal state (via snapshot) and any new output that arrives.
461
462        Args:
463            pattern: The regex pattern to look for in the terminal output
464            timeout: Maximum time to wait in seconds (default: 5.0)
465
466        Raises:
467            TimeoutError: If the pattern doesn't appear within the timeout period
468            RuntimeError: If the ht process has exited
469        """
470        if self._ht_proc.poll() is not None:
471            raise RuntimeError(f"ht process has exited with code {self._ht_proc.returncode}")
472
473        self._logger.debug(f"Expecting regex pattern: '{pattern}'")
474
475        # Compile the regex pattern
476        try:
477            regex = re.compile(pattern, re.MULTILINE)
478        except re.error as e:
479            raise ValueError(f"Invalid regex pattern '{pattern}': {e}") from e
480
481        # First check current terminal state
482        snapshot = self.snapshot()
483        if regex.search(snapshot.text):
484            self._logger.debug(f"Pattern '{pattern}' found immediately in current terminal state")
485            return
486
487        # Start time for timeout tracking
488        start_time = time.time()
489
490        # Process events until we find the pattern or timeout
491        while True:
492            # Check timeout
493            if time.time() - start_time > timeout:
494                self._logger.debug(f"Pattern '{pattern}' not found in terminal output after {timeout} seconds")
495                raise TimeoutError(f"Pattern '{pattern}' not found within {timeout} seconds")
496
497            try:
498                # Wait for next event with a short timeout to allow checking the overall timeout
499                event = self._event_queue.get(block=True, timeout=0.1)
500            except queue.Empty:
501                continue
502
503            # Process the event
504            if event["type"] == "output":
505                self._output_events.append(event)
506                # Check if pattern appears in this output
507                if "data" in event and "seq" in event["data"] and regex.search(event["data"]["seq"]):
508                    self._logger.debug(f"Pattern '{pattern}' found in output event")
509                    return
510            elif event["type"] == "exitCode":
511                # Put back in queue for reader thread to handle
512                self._event_queue.put(event)
513                # Don't raise here - the process might have exited after outputting what we want
514            elif event["type"] == "snapshot":
515                # If we get a snapshot event, check its content
516                if "data" in event and "text" in event["data"] and regex.search(event["data"]["text"]):
517                    self._logger.debug(f"Pattern '{pattern}' found in snapshot event")
518                    return
519
520            # Take a new snapshot periodically to catch any missed output
521            if len(self._output_events) % 10 == 0:  # Every 10 events
522                snapshot = self.snapshot()
523                if regex.search(snapshot.text):
524                    self._logger.debug(f"Pattern '{pattern}' found in periodic snapshot")
525                    return
526
527    def expect_absent(self, pattern: str, timeout: float = DEFAULT_EXPECT_TIMEOUT) -> None:
528        """
529        Wait for a regex pattern to disappear from the terminal output.
530
531        This method efficiently waits for output changes by monitoring the output events
532        from the ht process rather than polling with snapshots. It periodically checks
533        the terminal state to verify the pattern is gone.
534
535        Args:
536            pattern: The regex pattern that should disappear from the terminal output
537            timeout: Maximum time to wait in seconds (default: 5.0)
538
539        Raises:
540            TimeoutError: If the pattern doesn't disappear within the timeout period
541            RuntimeError: If the ht process has exited
542        """
543        if self._ht_proc.poll() is not None:
544            raise RuntimeError(f"ht process has exited with code {self._ht_proc.returncode}")
545
546        self._logger.debug(f"Expecting regex pattern to disappear: '{pattern}'")
547
548        # Compile the regex pattern
549        try:
550            regex = re.compile(pattern, re.MULTILINE)
551        except re.error as e:
552            raise ValueError(f"Invalid regex pattern '{pattern}': {e}") from e
553
554        # Start time for timeout tracking
555        start_time = time.time()
556
557        while True:
558            # Take a snapshot to check current state
559            snapshot = self.snapshot()
560            if not regex.search(snapshot.text):
561                self._logger.debug(f"Pattern '{pattern}' is now absent from terminal output")
562                return
563
564            # Check timeout
565            if time.time() - start_time > timeout:
566                self._logger.debug(f"Pattern '{pattern}' still present in terminal output after {timeout} seconds")
567                raise TimeoutError(f"Pattern '{pattern}' still present after {timeout} seconds")
568
569            # Wait for next event with a short timeout
570            try:
571                event = self._event_queue.get(block=True, timeout=0.1)
572            except queue.Empty:
573                continue
574
575            # Process the event
576            if event["type"] == "output":
577                self._output_events.append(event)
578            elif event["type"] == "exitCode":
579                # Put back in queue for reader thread to handle
580                self._event_queue.put(event)
581                # Don't raise here - the process might have exited after the pattern disappeared