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

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

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