511@contextmanager
512def terminal_session(
513    command: Command,
514    rows: Rows = None,
515    cols: Cols = None,
516    logger: Logger = None,
517    extra_subscribes: ExtraSubscribes = None,
518) -> Iterator[HtWrapper]:
519    """
520    The terminal_session context manager is a wrapper around `run` which ensures that the underlying process
521    gets cleaned up:
522
523    ```python
524    with terminal_session("some command") as proc:
525        # interact with the running command here
526        assert proc.exit_code is not None
527        s = proc.snapshot()
528
529    # htty terminates your command on context exit
530    assert proc.exit_code is None
531    assert "hello world" in proc.snapshot()
532    ```
533
534    Its usage is otherwise the same as `run`.
535    It also returns a
536
537    """
538
539    proc = run(
540        command,
541        rows=rows,
542        cols=cols,
543        no_exit=True,
544        logger=logger,
545        extra_subscribes=extra_subscribes,
546    )
547    try:
548        yield proc
549    finally:
550        try:
551            if proc.cmd.pid:
552                proc.cmd.terminate()
553                proc.cmd.wait(timeout=DEFAULT_SUBPROCESS_WAIT_TIMEOUT)
554        except Exception:
555            try:
556                if proc.cmd.pid:
557                    proc.cmd.kill()
558            except Exception:
559                pass
560
561        try:
562            proc.ht.terminate()
563            proc.ht.wait(timeout=DEFAULT_SUBPROCESS_WAIT_TIMEOUT)
564        except Exception:
565            with suppress(Exception):
566                proc.ht.kill()

569def run(
570    command: Command,
571    rows: Rows = None,
572    cols: Cols = None,
573    no_exit: NoExit = True,
574    logger: Logger = None,
575    extra_subscribes: ExtraSubscribes = None,
576) -> HtWrapper:
577    """
578    As a user of the htty python library, your code will run in the python process at the root of this
579    process tree:
580
581        python '{/path/to/your/code.py}'
582        └── ht
583            └── sh -c '{modified command}'
584
585    So if you're using htty to wrap vim, the process tree is:
586
587    ```
588    python
589    └── ht
590        └── sh
591            └── vim
592    ```
593
594    For reasons that are documented in [htty-core](./htty-core/htty_core.html#run), the command that ht
595    runs is not:
596
597        sh -c '{command}'
598
599    Instead it's something like this:
600
601        sh -c '{command} ; exit_code=$? ; /path/to/ht wait-exit /path/to/tmp/ht_fifo_5432 ; exit $exit_code'
602
603    This function invokes `ht` as a subprocess such that you end up with a process tree like the one shown
604    above. It returns an `HtWrapper` object which can be used to interact with ht and its child process.
605
606    It's up to you to clean up this process when you're done:
607
608    ```python
609    proc = run("some command")
610    # do stuff
611    proc.exit()
612    ````
613    If you'd rather not risk having a bunch of `ht` processes lying around and wasting CPU cycles,
614    consider using the `terminal_session` instead.
615    """
616    # Use provided logger or fall back to default
617    process_logger = logger or default_logger
618
619    # Create a queue for events
620    event_queue: queue.Queue[dict[str, Any]] = queue.Queue()
621
622    # Build the ht subscription list
623    base_subscribes = [
624        HtEvent.INIT,
625        HtEvent.SNAPSHOT,
626        HtEvent.OUTPUT,
627        HtEvent.RESIZE,
628        HtEvent.PID,
629        HtEvent.EXIT_CODE,
630        HtEvent.COMMAND_COMPLETED,
631    ]
632    if extra_subscribes:
633        # Convert string subscribes to HtEvent enum values
634        for sub in extra_subscribes:
635            try:
636                base_subscribes.append(HtEvent(sub))
637            except ValueError:
638                process_logger.warning(f"Unknown subscription event: {sub}")
639
640    # Convert command to string if it's a list, properly escaping shell arguments
641    command_str = command if isinstance(command, str) else " ".join(shlex.quote(arg) for arg in command)
642
643    # Create HtArgs and use htty_core.run()
644    ht_args = HtArgs(
645        command=command_str,  # Use the already-formatted command string
646        subscribes=base_subscribes,
647        rows=rows,
648        cols=cols,
649    )
650
651    # Log the exact command that would be run
652    cmd_args = ht_args.get_command()
653    process_logger.debug(f"Launching command: {' '.join(cmd_args)}")
654
655    ht_proc = htty_core_run(ht_args)
656
657    process_logger.debug(f"ht started: PID {ht_proc.pid}")
658
659    # Create a reader thread to capture ht output
660    def reader_thread(
661        ht_proc: subprocess.Popen[str],
662        queue_obj: queue.Queue[dict[str, Any]],
663        ht_process: HtWrapper,
664        thread_logger: logging.Logger,
665    ) -> None:
666        thread_logger.debug(f"Reader thread started for ht process {ht_proc.pid}")
667
668        while True:
669            if ht_proc.stdout is None:
670                thread_logger.warning(f"ht process {ht_proc.pid} stdout is None, exiting reader thread")
671                break
672
673            line = ht_proc.stdout.readline()
674            if not line:
675                thread_logger.debug(f"ht process {ht_proc.pid} stdout closed, exiting reader thread")
676                break
677
678            line = line.strip()
679            if not line:
680                continue
681
682            try:
683                event = json.loads(line)
684                thread_logger.debug(f"ht event: {event}")
685                queue_obj.put(event)
686
687                if event["type"] == "output":
688                    ht_process.add_output_event(event)
689                elif event["type"] == "exitCode":
690                    thread_logger.debug(
691                        f"ht process {ht_proc.pid} subprocess exited with code: {event.get('data', {}).get('exitCode')}"
692                    )
693                    ht_process.set_subprocess_exited(True)
694                    exit_code = event.get("data", {}).get("exitCode")
695                    if exit_code is not None:
696                        ht_process.cmd.exit_code = exit_code
697                elif event["type"] == "pid":
698                    thread_logger.debug(f"ht process {ht_proc.pid} subprocess PID: {event.get('data', {}).get('pid')}")
699                    pid = event.get("data", {}).get("pid")
700                    if pid is not None:
701                        ht_process.cmd.pid = pid
702                elif event["type"] == "commandCompleted":
703                    # Command has completed - this is the reliable signal that subprocess finished
704                    ht_process.set_subprocess_completed(True)
705                elif event["type"] == "debug":
706                    thread_logger.debug(f"ht process {ht_proc.pid} debug: {event.get('data', {})}")
707                    # Note: We no longer rely on debug events for subprocess_completed
708                    # The commandCompleted event (above) is the reliable source
709            except json.JSONDecodeError as e:
710                # Only log raw stdout when we can't parse it as JSON - this indicates an unexpected message
711                thread_logger.warning(f"ht process {ht_proc.pid} non-JSON stdout: {line} (error: {e})")
712                pass
713
714        thread_logger.debug(f"Reader thread exiting for ht process {ht_proc.pid}")
715
716    # Create an HtWrapper instance
717    process = HtWrapper(
718        ht_proc,
719        event_queue,
720        command=command_str,
721        rows=rows,
722        cols=cols,
723        no_exit=no_exit,
724        logger=process_logger,
725    )
726
727    # Start the reader thread for stdout
728    stdout_thread = threading.Thread(
729        target=reader_thread,
730        args=(ht_proc, event_queue, process, process_logger),
731        daemon=True,
732    )
733    stdout_thread.start()
734
735    # Start a stderr reader thread
736    def stderr_reader_thread(ht_proc: subprocess.Popen[str], thread_logger: logging.Logger) -> None:
737        thread_logger.debug(f"Stderr reader thread started for ht process {ht_proc.pid}")
738
739        while True:
740            if ht_proc.stderr is None:
741                thread_logger.warning(f"ht process {ht_proc.pid} stderr is None, exiting stderr reader thread")
742                break
743
744            line = ht_proc.stderr.readline()
745            if not line:
746                thread_logger.debug(f"ht process {ht_proc.pid} stderr closed, exiting stderr reader thread")
747                break
748
749            line = line.strip()
750            if line:
751                thread_logger.debug(f"ht stderr: {line}")
752
753        thread_logger.debug(f"Stderr reader thread exiting for ht process {ht_proc.pid}")
754
755    stderr_thread = threading.Thread(target=stderr_reader_thread, args=(ht_proc, process_logger), daemon=True)
756    stderr_thread.start()
757
758    # Wait briefly for the process to initialize and get PID
759    start_time = time.time()
760    while time.time() - start_time < 2:
761        try:
762            event = event_queue.get(block=True, timeout=0.5)
763            if event["type"] == "pid":
764                pid = event["data"]["pid"]
765                process.cmd.pid = pid
766                break
767        except queue.Empty:
768            continue
769
770    time.sleep(DEFAULT_SLEEP_AFTER_KEYS)
771    return process

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