Bump version to 2.3.0

Adds Esc-to-quit, working scroll (keys + mouse wheel), and click-to-jump
in the interactive filter UI.

CHANGELOG.md

@@ -1,6 +1,18 @@
 Change Log
 ==========
 
+Version 2.3.0 *(2026-07-20)*
+----------------------------
+
+ * New: Esc quits the interactive UI. A bare Escape keypress exits; Escape sequences
+   (arrow keys, mouse reports) are still parsed normally.
+ * Fix: Scrolling the interactive UI now works. Up/Down, Page Up/Down, Home/End, and
+   the mouse wheel scroll the retained scrollback; new lines no longer yank you back
+   to the tail while you're scrolled up.
+ * New: Clicking a line in the interactive UI clears the filter, unfilters the log,
+   and scrolls to that line in the full history.
+
+
 Version 2.2.0 *(2026-07-20)*
 ----------------------------
 

CLAUDE.md

@@ -0,0 +1,82 @@
+# pidcat-repl
+
+A fork of [JakeWharton/pidcat](https://github.com/JakeWharton/pidcat) that adds an
+interactive, full-screen filter UI on top of the original package-filtered `adb logcat`
+viewer.
+
+## Layout
+
+Everything lives in a single script, `pidcat.py`. There is no build step, no test suite,
+and no dependencies beyond Python 3 and `adb` on the PATH.
+
+- `stream(emit)` parses `adb logcat` output and hands each formatted block to a callback.
+  Both output modes feed off it.
+- `InteractiveUI` is the fork's addition: alternate-screen rendering, a bottom prompt line,
+  and live filtering of the retained scrollback.
+- `--plain` bypasses `InteractiveUI` and prints blocks as they arrive, which is the
+  upstream behavior. Non-TTY stdin or stdout forces plain mode too.
+
+Run it directly during development: `./pidcat.py com.example.app`.
+
+## Homebrew distribution
+
+This fork is distributed through a personal tap, **not** homebrew-core. Two repos are
+involved:
+
+- `imcarlost/pidcat-repl` (this repo) holds the source and the version tags.
+- `imcarlost/homebrew-pidcat-repl` holds `Formula/pidcat-repl.rb`. Cloned locally at
+  `../homebrew-pidcat-repl`.
+
+The core formula named `pidcat` is upstream's, unrelated to this one. Homebrew derives the
+formula class name from the filename, so `pidcat-repl.rb` must keep the class
+`PidcatRepl`.
+
+### Publishing a new version
+
+The formula pins a tarball URL plus its sha256, so the version bump, the tag, and the
+formula must all move together. Order matters: the sha256 can only be computed after the
+tag is pushed, because it hashes GitHub's generated tarball.
+
+1. Bump `__version__` in `pidcat.py` and promote the `Unreleased` heading in
+   `CHANGELOG.md` to the new version with today's date.
+2. Commit and push to `main`.
+3. Tag and push: `git tag vX.Y.Z && git push origin vX.Y.Z`.
+4. Compute the tarball hash:
+
+   ```sh
+   curl -sL https://github.com/imcarlost/pidcat-repl/archive/refs/tags/vX.Y.Z.tar.gz \
+     | shasum -a 256
+   ```
+
+5. In `../homebrew-pidcat-repl/Formula/pidcat-repl.rb`, update both the `url` version and
+   the `sha256`. Commit and push.
+
+Keep `__version__` in sync with the tag. A mismatch is invisible until someone runs
+`pidcat --version` and sees the old number.
+
+If a tag has to be moved after it was pushed, the tarball hash changes with it, so the
+formula's `sha256` must be recomputed and pushed again or installs fail checksum
+verification.
+
+### Installing and testing the tap
+
+```sh
+brew tap imcarlost/pidcat-repl
+brew install pidcat-repl
+```
+
+Recent Homebrew versions refuse formulae from untrusted third-party taps. The install
+fails with a `Refusing to load formula ... from untrusted tap` error until the tap is
+trusted once per machine:
+
+```sh
+brew trust imcarlost/pidcat-repl
+```
+
+That is a local trust decision each user makes for themselves; there is no way to
+pre-authorize the tap from the publishing side. Getting into homebrew-core instead is not
+realistic for this fork, since core already carries upstream `pidcat` and forks rarely
+meet its notability bar.
+
+After pushing a formula change, `brew update` before reinstalling, or the tap clone under
+`$(brew --repository)/Library/Taps/` will still be on the old commit.

README.md

@@ -48,22 +48,13 @@ uses plain streaming, since there is no terminal to draw the UI on.
 Install
 -------
 
-This fork isn't packaged anywhere; download `pidcat.py` from this repo and
+Use [Homebrew][2]:
-place it on your PATH.
 
-The Homebrew and AUR packages below install the upstream, non-interactive
+```shell
-`pidcat` instead:
+brew tap imcarlost/pidcat-repl
-
+brew trust imcarlost/pidcat-repl
- *  OS X: Use [Homebrew][2].
+brew install pidcat-repl
-
+```
-         brew install pidcat
-
-    If you need to install the latest development version
-
-        brew unlink pidcat
-        brew install --HEAD pidcat
-
- * Arch Linux : Install the package called `pidcat-git` from the [AUR][4].
 
 
 Make sure that `adb` from the [Android SDK][3] is on your PATH. This script will
@@ -92,5 +83,4 @@ sudo apt-get -t focal install coreutils
  [1]: http://jsharkey.org/blog/2009/04/22/modifying-the-android-logcat-stream-for-full-color-debugging/
  [2]: http://brew.sh
  [3]: http://developer.android.com/sdk/
- [4]: https://aur.archlinux.org/packages/pidcat-git/
  [5]: https://github.com/JakeWharton/pidcat

pidcat.py

@@ -33,7 +33,7 @@ import subprocess
 import threading
 from subprocess import PIPE
 
-__version__ = '2.2.0'
+__version__ = '2.3.0'
 
 LOG_LEVELS = 'VDIWEF'
 LOG_LEVELS_MAP = dict([(LOG_LEVELS[i], i) for i in range(len(LOG_LEVELS))])
@@ -392,11 +392,14 @@ class InteractiveUI:
 
   def __init__(self):
     self.lock = threading.Lock()
-    self.entries = collections.deque(maxlen=self.MAX_ENTRIES)
+    self.entries = collections.deque(maxlen=self.MAX_ENTRIES)  # (entry_id, search_lower, block)
-    self.visible = []
+    self.visible = []  # (entry_id, line_text)
+    self.next_entry_id = 0
     self.query = ''
     self.status = ''
     self.resized = False
+    self.scroll_offset = 0  # lines scrolled up from the tail; 0 == following the live tail
+    self.render_start = 0   # index into self.visible of the top rendered line
     self.rows, self.cols = self._term_size()
 
   def _term_size(self):
@@ -412,26 +415,36 @@ class InteractiveUI:
   def _matches(self, search_lower):
     return all(token in search_lower for token in self.query.lower().split())
 
-  def _append_visible(self, block):
+  def _append_visible(self, entry_id, block):
-    self.visible.extend(block.split('\n'))
+    new_lines = [(entry_id, line) for line in block.split('\n')]
-    if len(self.visible) > self.MAX_VISIBLE:
+    self.visible.extend(new_lines)
-      del self.visible[:len(self.visible) - self.MAX_VISIBLE]
+    if self.scroll_offset > 0:
+      # Keep whatever the user is currently looking at in place instead of
+      # letting newly-arrived lines push it down and out of view.
+      self.scroll_offset += len(new_lines)
+    overflow = len(self.visible) - self.MAX_VISIBLE
+    if overflow > 0:
+      del self.visible[:overflow]
+      self.scroll_offset = max(0, self.scroll_offset - overflow)
 
   def emit(self, search_text, block):
     with self.lock:
+      entry_id = self.next_entry_id
+      self.next_entry_id += 1
       search_lower = search_text.lower()
-      self.entries.append((search_lower, block))
+      self.entries.append((entry_id, search_lower, block))
       if self._matches(search_lower):
-        self._append_visible(block)
+        self._append_visible(entry_id, block)
       self._render()
 
   def set_query(self, query):
     with self.lock:
       self.query = query
       self.visible = []
-      for search_lower, block in self.entries:
+      self.scroll_offset = 0
+      for entry_id, search_lower, block in self.entries:
         if self._matches(search_lower):
-          self._append_visible(block)
+          self._append_visible(entry_id, block)
       self._render()
 
   def refresh(self):
@@ -443,40 +456,149 @@ class InteractiveUI:
 
   def _render(self):
     log_rows = max(1, self.rows - 2)
-    lines = self.visible[-log_rows:]
+    max_offset = max(0, len(self.visible) - log_rows)
+    if self.scroll_offset > max_offset:
+      self.scroll_offset = max_offset
+    end = len(self.visible) - self.scroll_offset
+    start = max(0, end - log_rows)
+    self.render_start = start
+    window = self.visible[start:end]
     out = ['\x1b[H']
     # Pad above so the log content hugs the prompt, like a terminal.
-    for _ in range(log_rows - len(lines)):
+    for _ in range(log_rows - len(window)):
       out.append('\x1b[K\n')
-    for line in lines:
+    for _, line in window:
       out.append(line + '\x1b[K\n')
     if self.query:
       state = '%d matching lines of %d blocks' % (len(self.visible), len(self.entries))
     else:
       state = '%d blocks' % len(self.entries)
+    if self.scroll_offset > 0:
+      state += ' \xb7 scrolled (End to jump to latest)'
     if self.status:
       state += ' \xb7 ' + self.status
-    separator = ' %s \xb7 type to filter \xb7 ctrl-u clear \xb7 ctrl-c quit' % state
+    separator = ' %s \xb7 type to filter \xb7 ctrl-u clear \xb7 click a line to unfilter \xb7 esc/ctrl-c quit' % state
     out.append('\x1b[2m' + separator[:max(0, self.cols - 1)] + '\x1b[0m\x1b[K\n')
     out.append('\x1b[36m❯\x1b[0m ' + self.query + '\x1b[K')
     sys.stdout.write(''.join(out))
     sys.stdout.flush()
 
-  def _drain_pending(self, fd):
+  MOUSE_RE = re.compile(r'^\[<(\d+);(\d+);(\d+)([Mm])$')
-    # Swallow the rest of an escape sequence (arrow keys etc.) we do not handle.
+
-    while select.select([fd], [], [], 0)[0]:
+  def _scroll(self, delta):
-      if not os.read(fd, 64):
+    with self.lock:
+      log_rows = max(1, self.rows - 2)
+      max_offset = max(0, len(self.visible) - log_rows)
+      self.scroll_offset = max(0, min(max_offset, self.scroll_offset + delta))
+      self._render()
+
+  def _handle_click(self, row):
+    with self.lock:
+      log_rows = max(1, self.rows - 2)
+      if row < 1 or row > log_rows:
+        return
+      idx = self.render_start + (row - 1)
+      if idx < 0 or idx >= len(self.visible):
+        return
+      entry_id = self.visible[idx][0]
+    self._jump_to_entry(entry_id)
+
+  def _jump_to_entry(self, entry_id):
+    '''Clears the filter, rebuilds the full unfiltered scrollback, and scrolls
+    so the clicked entry is in view with a bit of context above it.'''
+    with self.lock:
+      self.query = ''
+      self.visible = []
+      self.scroll_offset = 0
+      target_index = None
+      for eid, _search_lower, block in self.entries:
+        for line in block.split('\n'):
+          if target_index is None and eid == entry_id:
+            target_index = len(self.visible)
+          self.visible.append((eid, line))
+      overflow = len(self.visible) - self.MAX_VISIBLE
+      if overflow > 0:
+        del self.visible[:overflow]
+        if target_index is not None:
+          target_index -= overflow
+      if target_index is not None and target_index >= 0:
+        log_rows = max(1, self.rows - 2)
+        max_offset = max(0, len(self.visible) - log_rows)
+        desired_end = target_index + max(1, log_rows // 3)
+        self.scroll_offset = max(0, min(max_offset, len(self.visible) - desired_end))
+      self._render()
+
+  def _handle_escape(self, seq):
+    m = self.MOUSE_RE.match(seq)
+    if m:
+      button, _col, row, kind = m.groups()
+      if kind != 'M':  # ignore button-release reports
+        return
+      button = int(button)
+      if button == 0:  # left click
+        self._handle_click(int(row))
+      elif button == 64:  # wheel up
+        self._scroll(3)
+      elif button == 65:  # wheel down
+        self._scroll(-3)
+      return
+    log_rows = max(1, self.rows - 2)
+    if seq in ('[A', 'OA'):  # up
+      self._scroll(1)
+    elif seq in ('[B', 'OB'):  # down
+      self._scroll(-1)
+    elif seq == '[5~':  # page up
+      self._scroll(log_rows)
+    elif seq == '[6~':  # page down
+      self._scroll(-log_rows)
+    elif seq in ('[H', '[1~'):  # home
+      self._scroll(10 ** 9)
+    elif seq in ('[F', '[4~'):  # end
+      self._scroll(-(10 ** 9))
+
+  def _read_escape(self, pending, fd, decoder):
+    '''Consumes the sequence following an ESC already pulled from `pending`.
+    Returns (seq, rest_of_pending). seq is None for a standalone Escape
+    keypress (nothing followed it within the grace window), '' when ESC was
+    followed by something unrelated, or the sequence body (e.g. '[A')
+    otherwise.'''
+    if not pending:
+      # Give a fast terminal-generated sequence (arrow keys, mouse reports)
+      # a brief window to arrive before treating this as a lone Escape.
+      if select.select([fd], [], [], 0.01)[0]:
+        data = os.read(fd, 64)
+        if data:
+          pending = decoder.decode(data)
+    if not pending:
+      return None, pending
+    introducer = pending[0]
+    if introducer not in ('[', 'O'):
+      return '', pending
+    seq = introducer
+    pending = pending[1:]
+    while not (seq[-1].isalpha() or seq[-1] == '~'):
+      if not pending:
+        if select.select([fd], [], [], 0.01)[0]:
+          data = os.read(fd, 64)
+          if data:
+            pending += decoder.decode(data)
+            continue
         break
+      seq += pending[0]
+      pending = pending[1:]
+    return seq, pending
 
   def run(self, reader_thread):
     fd = sys.stdin.fileno()
     old_attrs = termios.tcgetattr(fd)
     decoder = codecs.getincrementaldecoder('utf-8')('replace')
     signal.signal(signal.SIGWINCH, lambda *_: setattr(self, 'resized', True))
-    sys.stdout.write('\x1b[?1049h\x1b[?7l\x1b[2J\x1b[H')  # alt screen, no autowrap
+    # Alt screen, no autowrap, mouse click/wheel reporting (SGR encoding).
+    sys.stdout.write('\x1b[?1049h\x1b[?7l\x1b[2J\x1b[H\x1b[?1000h\x1b[?1006h')
     sys.stdout.flush()
     tty.setcbreak(fd)
     reader_thread.start()
+    pending = ''  # decoded characters carried over between reads
     try:
       with self.lock:
         self._render()
@@ -488,35 +610,41 @@ class InteractiveUI:
           self.status = 'adb ended, scrollback still filterable'
           with self.lock:
             self._render()
-        if not select.select([fd], [], [], 0.2)[0]:
+        if not pending:
-          continue
+          if not select.select([fd], [], [], 0.2)[0]:
-        data = os.read(fd, 64)
+            continue
-        if not data:
+          data = os.read(fd, 64)
-          break
+          if not data:
-        for ch in decoder.decode(data):
-          if ch in ('\x7f', '\x08'):  # backspace
-            if self.query:
-              self.set_query(self.query[:-1])
-          elif ch == '\x15':  # ctrl-u
-            if self.query:
-              self.set_query('')
-          elif ch == '\x04':  # ctrl-d
-            return
-          elif ch == '\x0c':  # ctrl-l
-            self.refresh()
-          elif ch == '\x1b':
-            decoder.reset()
-            self._drain_pending(fd)
             break
-          elif ch in ('\r', '\n', '\t'):
+          pending = decoder.decode(data)
-            pass
+          if not pending:
-          elif ch >= ' ':
+            continue
-            self.set_query(self.query + ch)
+        ch, pending = pending[0], pending[1:]
+        if ch in ('\x7f', '\x08'):  # backspace
+          if self.query:
+            self.set_query(self.query[:-1])
+        elif ch == '\x15':  # ctrl-u
+          if self.query:
+            self.set_query('')
+        elif ch == '\x04':  # ctrl-d
+          return
+        elif ch == '\x0c':  # ctrl-l
+          self.refresh()
+        elif ch == '\x1b':
+          seq, pending = self._read_escape(pending, fd, decoder)
+          if seq is None:  # standalone Escape keypress
+            return
+          if seq:
+            self._handle_escape(seq)
+        elif ch in ('\r', '\n', '\t'):
+          pass
+        elif ch >= ' ':
+          self.set_query(self.query + ch)
     except KeyboardInterrupt:
       pass
     finally:
       termios.tcsetattr(fd, termios.TCSADRAIN, old_attrs)
-      sys.stdout.write('\x1b[?7h\x1b[?1049l')
+      sys.stdout.write('\x1b[?1006l\x1b[?1000l\x1b[?7h\x1b[?1049l')
       sys.stdout.flush()